import os import uuid import sqlite3 import shutil import json import requests as _requests from urllib.parse import urlparse as _urlparse from dotenv import load_dotenv from flask import request, render_template, redirect, url_for, jsonify, session, flash, abort, g, send_from_directory load_dotenv() # Read website_url directly from .env — single source of truth website_url = os.getenv("website_url", "localhost") from werkzeug.utils import secure_filename from helpers import _db, url, get_site_settings, get_text_overrides, dir_path, get_client_ip, check_conversion_limit, record_tool_usage, get_current_user, get_plan_by_id, send_job_notification_email, get_vapid_keys, send_push_notification, get_page_content, log_deletion from routes.subdomain import _filetype_from_host, _is_subdomains_enabled from configs.filetypes import available_filetypes, available_hashtypes from configs.definition import definitions from configs.languages import supported_languages def get_user_batch_limit(): """Return the current user's max batch file count from their plan. Returns 9999 for unlimited (plan value 0), falls back to 20 (DB column default) when the column is absent, null, or the plan cannot be read. """ try: _user = get_current_user() _plan = get_plan_by_id(_user['plan_id']) if _user else get_plan_by_id(1) if _plan and _plan.get('max_batch_files') is not None: _raw = int(_plan['max_batch_files']) return 9999 if _raw == 0 else (_raw if _raw > 0 else 20) except Exception: pass return 20 PDF_TOOL_META = { "merge-pdf": {"title": "Merge PDF", "desc": "Merge multiple PDF files into a single PDF document", "icon": "fa-object-group", "accept": ".pdf", "multi": True, "category": "organize"}, "split-pdf": {"title": "Split PDF", "desc": "Split PDF into multiple files by page ranges", "icon": "fa-code-branch", "accept": ".pdf", "multi": False, "category": "organize"}, "rotate-pdf": {"title": "Rotate PDF", "desc": "Rotate individual pages or entire PDF document", "icon": "fa-rotate", "accept": ".pdf", "multi": False, "category": "organize"}, "remove-pages": {"title": "Remove Pages", "desc": "Delete unwanted pages from your PDF file", "icon": "fa-file-circle-minus", "accept": ".pdf", "multi": False, "category": "organize"}, "organize-pdf": {"title": "Organize PDF", "desc": "Reorder, delete, and organize pages visually", "icon": "fa-layer-group", "accept": ".pdf", "multi": True, "category": "organize"}, "extract-pages": {"title": "Extract Pages", "desc": "Extract selected pages into a new PDF document", "icon": "fa-scissors", "accept": ".pdf", "multi": False, "category": "organize"}, "compress-pdf": {"title": "Compress PDF", "desc": "Reduce PDF file size with adjustable compression levels", "icon": "fa-compress", "accept": ".pdf", "multi": False, "category": "optimize"}, "optimize-pdf": {"title": "Optimize PDF", "desc": "Optimize PDF for web, print, or ebook", "icon": "fa-gauge-high", "accept": ".pdf", "multi": False, "category": "optimize"}, "grayscale-pdf": {"title": "Grayscale PDF", "desc": "Convert all colors to grayscale for printing", "icon": "fa-palette", "accept": ".pdf", "multi": False, "category": "convert"}, "resize-pdf": {"title": "Resize PDF", "desc": "Change page size and scale content", "icon": "fa-expand", "accept": ".pdf", "multi": False, "category": "convert"}, "crop-pdf": {"title": "Crop PDF", "desc": "Crop margins and remove white space", "icon": "fa-crop", "accept": ".pdf", "multi": False, "category": "convert"}, "repair-pdf": {"title": "Repair PDF", "desc": "Fix corrupted or damaged PDF files", "icon": "fa-wrench", "accept": ".pdf", "multi": False, "category": "tools"}, "protect-pdf": {"title": "Protect PDF", "desc": "Add password encryption with permissions", "icon": "fa-lock", "accept": ".pdf", "multi": False, "category": "security"}, "unlock-pdf": {"title": "Unlock PDF", "desc": "Remove password protection from PDF files", "icon": "fa-unlock", "accept": ".pdf", "multi": False, "category": "security"}, "sign-pdf": {"title": "Sign PDF", "desc": "Add digital signature to your PDF", "icon": "fa-pen", "accept": ".pdf", "multi": False, "category": "security"}, "watermark-pdf": {"title": "Watermark PDF", "desc": "Add text or image watermark to PDF", "icon": "fa-droplet", "accept": ".pdf", "multi": False, "category": "annotate"}, "add-page-numbers": {"title": "Add Page Numbers", "desc": "Insert page numbers with custom formatting", "icon": "fa-hashtag", "accept": ".pdf", "multi": False, "category": "annotate"}, "flatten-pdf": {"title": "Flatten PDF", "desc": "Flatten annotations and form fields", "icon": "fa-layer-group", "accept": ".pdf", "multi": False, "category": "convert"}, "pdf-to-jpg": {"title": "PDF to JPG", "desc": "Convert PDF pages to high-quality JPG images", "icon": "fa-image", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-png": {"title": "PDF to PNG", "desc": "Convert PDF pages to transparent PNG images", "icon": "fa-image", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-bmp": {"title": "PDF to BMP", "desc": "Convert PDF pages to BMP bitmap images", "icon": "fa-image", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-tiff": {"title": "PDF to TIFF", "desc": "Convert PDF to multi-page TIFF format", "icon": "fa-images", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-word": {"title": "PDF to Word", "desc": "Convert PDF to editable Word documents", "icon": "fa-file-word", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-powerpoint": {"title": "PDF to PowerPoint", "desc": "Convert PDF to PowerPoint presentations", "icon": "fa-file-powerpoint", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-excel": {"title": "PDF to Excel", "desc": "Extract tables from PDF to Excel format", "icon": "fa-file-excel", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-txt": {"title": "PDF to TXT", "desc": "Extract text content from PDF", "icon": "fa-file-lines", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-html": {"title": "PDF to HTML", "desc": "Convert PDF to web-friendly HTML", "icon": "fa-code", "accept": ".pdf", "multi": False, "category": "export"}, "pdf-to-zip": {"title": "PDF to ZIP", "desc": "Package PDF pages as ZIP archive", "icon": "fa-file-zipper", "accept": ".pdf", "multi": False, "category": "export"}, "jpg-to-pdf": {"title": "JPG to PDF", "desc": "Convert JPG images to PDF", "icon": "fa-file-image", "accept": ".jpg,.jpeg,.png,.bmp,.gif,.tiff", "multi": True, "category": "import"}, "png-to-pdf": {"title": "PNG to PDF", "desc": "Convert PNG images to PDF", "icon": "fa-file-image", "accept": ".png,.jpg,.jpeg,.bmp,.gif,.tiff", "multi": True, "category": "import"}, "bmp-to-pdf": {"title": "BMP to PDF", "desc": "Convert BMP images to PDF", "icon": "fa-file-image", "accept": ".bmp,.jpg,.jpeg,.png,.gif,.tiff", "multi": True, "category": "import"}, "tiff-to-pdf": {"title": "TIFF to PDF", "desc": "Convert TIFF images to PDF", "icon": "fa-images", "accept": ".tiff,.tif,.jpg,.jpeg,.png,.bmp", "multi": True, "category": "import"}, "gif-to-pdf": {"title": "GIF to PDF", "desc": "Convert GIF images to PDF", "icon": "fa-file-image", "accept": ".gif,.jpg,.jpeg,.png,.bmp,.tiff", "multi": True, "category": "import"}, "word-to-pdf": {"title": "Word to PDF", "desc": "Convert Word documents to PDF", "icon": "fa-file-word", "accept": ".doc,.docx", "multi": True, "category": "import"}, "powerpoint-to-pdf": {"title": "PowerPoint to PDF", "desc": "Convert PowerPoint to PDF", "icon": "fa-file-powerpoint", "accept": ".ppt,.pptx", "multi": True, "category": "import"}, "excel-to-pdf": {"title": "Excel to PDF", "desc": "Convert Excel spreadsheets to PDF", "icon": "fa-file-excel", "accept": ".xls,.xlsx", "multi": True, "category": "import"}, "txt-to-pdf": {"title": "TXT to PDF", "desc": "Convert text files to PDF", "icon": "fa-file-lines", "accept": ".txt", "multi": True, "category": "import"}, "html-to-pdf": {"title": "HTML to PDF", "desc": "Convert HTML to PDF", "icon": "fa-code", "accept": ".html,.htm", "multi": True, "category": "import"}, } from converters import image_converter, hash_generator, \ audio_converter, video_converter, \ document_converter, archive_converter, \ ebook_converter, device_converter, webservice_converter, \ pdf_converter, documents_compressor, image_compressor, \ video_compressor try: from rq import Queue from rq.job import Job from worker import conn as _redis_conn except Exception: Queue = None Job = None _redis_conn = None try: _redis_conn = _redis_conn except Exception: _redis_conn = None q = Queue(connection=_redis_conn) if (Queue and _redis_conn) else None _QUEUE_NAMES = ('high', 'medium', 'low') # _q_map is intentionally empty: there is no RQ worker process consuming Redis queues. # All jobs are processed by _local_queue (in-process background thread). # Redis / Job.fetch() is kept ONLY as a best-effort lookup for legacy job IDs. _q_map = {} from local_jobs import LocalJob, LocalQueue, _local_queue convert_list = { "image": image_converter, "bmp": image_converter, "hash": hash_generator, "audio": audio_converter, "video": video_converter, "document": document_converter, "archive": archive_converter, "ebook": ebook_converter, "device": device_converter, "webservice": webservice_converter, "pdf": pdf_converter, "document-compressor": documents_compressor, "image-compressor": image_compressor, "video-compressor": video_compressor, } _OUTPUT_ALLOWED = [ "image", "audio", "video", "document", "device", "archive", "ebook", "webservice", "pdf", "bmp", "hash", "document-compressor", "image-compressor", "video-compressor", ] def _gettext(string, **variables): from flask_babel import gettext as _original_gettext overrides = get_text_overrides() if string in overrides and overrides[string]: s = overrides[string] return s % variables if variables else s return _original_gettext(string, **variables) _ = _gettext def register_converter_routes(app): # ===== HELPER FUNCTIONS ===== def _sync_job_to_db(job, job_id): """Sync a finished/failed job's status into the conversions DB row.""" try: if job.is_finished or job.is_failed: new_status = 'finished' if job.is_finished else 'failed' _out_path = '' _err_msg = '' if job.result: r = job.result if isinstance(r, dict): results = r.get('results') or [] _out_path = (r.get('output_path') or (results[0] if results else '') or r.get('path') or r.get('file_path') or '') if r.get('error'): _err_msg = r.get('message', '') new_status = 'failed' elif isinstance(r, str): _out_path = r with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as _sc: _sc.execute( "UPDATE conversions SET status=?, " "output_path=COALESCE(NULLIF(?,''),output_path), " "error_message=CASE WHEN ? != '' THEN ? ELSE error_message END, " "completed_at=COALESCE(completed_at,datetime('now')) " "WHERE job_id=? AND status IN ('pending','started')", (new_status, _out_path, _err_msg, _err_msg, job_id) ) if _out_path: _sc.execute( "UPDATE conversions SET output_path=? WHERE job_id=? AND (output_path IS NULL OR output_path='')", (_out_path, job_id) ) _sc.commit() except Exception: pass def output(job_id, lang_code=None): """Handle output page for conversion results""" _ss = get_site_settings() _user = get_current_user() _plan = get_plan_by_id(_user['plan_id']) if _user else get_plan_by_id(1) _browser_notif_allowed = _ss.get('job_notify_browser_enabled', '1') != '0' _email_notif_allowed = ( _ss.get('job_notify_email_enabled', '0') == '1' and _plan and int(_plan.get('job_notify_email', 0)) ) _push_notif_allowed = _ss.get('job_notify_push_enabled', '0') == '1' _retry_available = False _conv_row = None try: _db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(_db_path) as _oc: _oc.row_factory = sqlite3.Row _conv_row = _oc.execute( "SELECT c.status, c.output_format, c.input_format, c.tool_type, " "c.user_id AS conv_user_id, c.ip_address AS conv_ip, " "fu.file_path AS upload_file_path " "FROM conversions c " "LEFT JOIN file_uploads fu ON fu.id = c.upload_id " "WHERE c.job_id=? ORDER BY c.id DESC LIMIT 1", (job_id,) ).fetchone() if _conv_row and _conv_row['status'] == 'failed': _upload_fp = _conv_row['upload_file_path'] if _upload_fp: _full = os.path.join(app.config.get('UPLOAD_DIR', 'storage/uploads'), _upload_fp) _retry_available = os.path.exists(_full) except Exception: _retry_available = False _upload_available = False if _conv_row and _conv_row['upload_file_path']: _uf = _conv_row['upload_file_path'] _uf_full = _uf if os.path.isabs(_uf) else os.path.join( app.config.get('UPLOAD_DIR', 'storage/uploads'), _uf) _upload_available = os.path.exists(_uf_full) _output_format = '' _input_format = '' _tool_type = '' if _conv_row: try: _output_format = _conv_row['output_format'] or '' _input_format = _conv_row['input_format'] or '' _tool_type = _conv_row['tool_type'] or '' except Exception: pass _conversion_tools = [] if _output_format: for _ft_name, _ft_cfg in available_filetypes.items(): _allowed_lower = [x.lower() for x in _ft_cfg.get('allowed', [])] if _output_format.lower() in _allowed_lower: _conversion_tools.append({ 'category': _ft_name, 'title': str(_ft_cfg.get('title', _ft_name.title() + ' Converter')), 'ext': [x.upper() for x in _ft_cfg.get('ext', [])], 'icon': _ft_cfg.get('icon-class', 'fa fa-file'), }) _tpl_kwargs = dict( job_id=job_id, browser_notif_allowed=_browser_notif_allowed, email_notif_allowed=_email_notif_allowed, push_notif_allowed=_push_notif_allowed, retry_available=_retry_available, upload_available=_upload_available, plan=_plan, output_format=_output_format, input_format=_input_format, tool_type=_tool_type, site_settings=_ss, conversion_tools=_conversion_tools, subdomains_enabled=_is_subdomains_enabled(), base_domain=(app.config.get('SERVER_NAME') or website_url), ) # ---- Access control: determine if visitor may view this output ---- # Fail-closed: default is to require permission. Only cleared when # ownership or a grant is positively confirmed. Exceptions keep # the gate closed (logged, not silently swallowed to allow access). _permission_required = True if not _conv_row: # No conversion record yet (job still queued/processing). # The processing page shows no sensitive file content, so allow. _permission_required = False else: try: from helpers import is_admin as _is_admin _conv_owner_id = _conv_row['conv_user_id'] _conv_ip = (_conv_row['conv_ip'] or '').strip() if _is_admin(): _permission_required = False elif _user and _conv_owner_id and _user['id'] == int(_conv_owner_id): # Logged-in user matches the recorded owner _permission_required = False elif not _conv_owner_id and _conv_ip and get_client_ip() == _conv_ip: # Guest-created job: same IP grants access regardless of login state _permission_required = False else: # Not positively identified as owner — check explicit access grants if _user: _user_email = (_user.get('email') or '').strip().lower() _user_id = _user.get('id') with sqlite3.connect(_db_path) as _gc: _gc.row_factory = sqlite3.Row _grant = _gc.execute( "SELECT 1 FROM job_access_grants " "WHERE job_id=? AND (LOWER(granted_to_email)=? OR granted_to_user_id=?) LIMIT 1", (job_id, _user_email, _user_id) ).fetchone() if _grant is not None: _permission_required = False # Back-fill user_id on the grant so future checks can use it try: _gc.execute( "UPDATE job_access_grants SET granted_to_user_id=? " "WHERE job_id=? AND LOWER(granted_to_email)=? AND granted_to_user_id IS NULL", (_user_id, job_id, _user_email) ) _gc.commit() except Exception: pass # Non-owner guests and ungranted users remain blocked except Exception as _authz_exc: app.logger.warning( f"Access control check failed for job {job_id}: {_authz_exc}" ) # Remain fail-closed: _permission_required stays True _tpl_kwargs['permission_required'] = _permission_required try: job = LocalJob.fetch(job_id) _sync_job_to_db(job, job_id) if _conv_row is None and (job.is_finished or job.is_failed): abort(404) if _conv_row and _conv_row['status'] == 'expired': abort(404) if _conv_row and _conv_row['status'] == 'deleted': _tpl_kwargs['conversion_deleted'] = True _tpl_kwargs['sanitized_results'] = [] return render_template('output.html', job=job, **_tpl_kwargs) import json as _json _sanitized_results = [] if job.is_finished and job.result and not job.result.get('error'): _upload_dir = app.config.get('UPLOAD_DIR', os.path.join(dir_path, 'storage', 'uploads')) for _r in (job.result.get('results') or []): if not _r: continue if str(_r).startswith('http'): _sanitized_results.append(_r) else: _full = _r if os.path.isabs(_r) else os.path.join(_upload_dir, _r) if os.path.isfile(_full): _sanitized_results.append(_r) if len(_sanitized_results) < len(job.result.get('results') or []): try: _upd_result = dict(job.result) _upd_result['results'] = _sanitized_results if not _sanitized_results: _upd_result['output_deleted'] = True with sqlite3.connect(db_path) as _upd: _upd.execute( "UPDATE local_jobs SET result_json=? WHERE id=?", (_json.dumps(_upd_result), job_id) ) job = LocalJob.fetch(job_id) except Exception: pass elif job.is_finished and job.result and job.result.get('output_deleted'): _sanitized_results = [] else: _sanitized_results = list(job.result.get('results') or []) if job.result else [] # If access is restricted, clear file URLs so they are never # rendered into the template JS (window._oc_output.file_urls). if _permission_required: _sanitized_results = [] _tpl_kwargs['sanitized_results'] = _sanitized_results return render_template('output.html', job=job, **_tpl_kwargs) except Exception: abort(404) def hashoutput(job_id, lang_code=None): job = None try: job = LocalJob.fetch(job_id) except Exception: pass if job is None: abort(404) # Initialize default values results2 = [] algorithm_name = "Hash" algorithm = "hash" supports_hmac = False supports_salt = False error = None try: # Check if job has a result if job.result: result = job.result if isinstance(result, dict): # Nested structure from jobStatus wrapper if 'result' in result and isinstance(result['result'], dict): result_data = result['result'] results2 = result_data.get('results2', []) else: results2 = result.get('results2', []) if result.get('is_failed', False) or result.get('error', False): error = {'message': result.get('message', 'Job failed')} # Metadata (Redis jobs only) if hasattr(job, 'meta') and job.meta: algorithm_name = job.meta.get('algorithm', 'Hash') algorithm = job.meta.get('algorithm_slug', algorithm_name.lower()) supports_hmac = job.meta.get('supports_hmac', False) supports_salt = job.meta.get('supports_salt', False) app.logger.info(f"Hash output for job {job_id}: results2={results2}") return render_template( 'output-hash.html', job=job, job_id=job_id, results2=results2, algorithm_name=algorithm_name, algorithm=algorithm, supports_hmac=supports_hmac, supports_salt=supports_salt, error=error ) except Exception as e: app.logger.error(f"Error in hashoutput for job {job_id}: {str(e)}") abort(404) def jobStatus(job_id): def _job_to_json(job): resp = { 'is_finished': job.is_finished, 'is_failed': job.is_failed, 'is_queued': job.is_queued, 'is_started': job.is_started, 'result': job.result } if job.is_finished or job.is_failed: import threading _sync_job_to_db(job, job_id) is_failed = job.is_failed threading.Thread(target=_trigger_job_email_notifications, args=(job_id, is_failed), daemon=True).start() threading.Thread(target=_trigger_job_push_notifications, args=(job_id, is_failed), daemon=True).start() return jsonify(resp) # Always try LocalJob first — all new jobs are enqueued in-process. # Fall back to Redis only for legacy job IDs that pre-date this change. try: job = LocalJob.fetch(job_id) return _job_to_json(job) except Exception: pass if Job and _redis_conn: try: job = Job.fetch(job_id, connection=_redis_conn) return _job_to_json(job) except Exception: pass abort(404) _job_email_locks = {} _job_email_locks_mu = __import__('threading').Lock() def _trigger_job_email_notifications(job_id, is_failed=False): with _job_email_locks_mu: if job_id in _job_email_locks: return _job_email_locks[job_id] = True try: ss = get_site_settings() if ss.get('job_notify_email_enabled', '0') != '1': return db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT id, email FROM job_notify_emails WHERE job_id=? AND sent=0", (job_id,) ).fetchall() if not rows: return claimed_ids = [r['id'] for r in rows] conn.execute( "UPDATE job_notify_emails SET sent=2 WHERE id IN ({})".format( ','.join('?' * len(claimed_ids)) ), claimed_ids ) conn.commit() site_name = ss.get('smtp_from_name', 'OnlineConvert') or 'OnlineConvert' for row in rows: if is_failed: subject = f"{site_name} – Conversion Failed" html = f"""

Conversion Failed

Unfortunately your conversion job {job_id[:8]}… encountered an error and could not be completed.

Please try again or contact support if the issue persists.

You received this because you subscribed to job notifications on {site_name}.

""" else: subject = f"{site_name} – Your conversion is ready!" html = f"""

Your file conversion is complete

Your conversion job {job_id[:8]}… has finished successfully.

Download Files

You received this because you subscribed to job notifications on {site_name}.

""" ok, _ = send_job_notification_email(row['email'], subject, html) conn.execute("UPDATE job_notify_emails SET sent=? WHERE id=?", (1 if ok else 0, row['id'],)) conn.commit() except Exception as e: app.logger.error(f"Job email notify error for {job_id}: {e}") finally: with _job_email_locks_mu: _job_email_locks.pop(job_id, None) @app.route('/convert/cancel/', methods=['POST']) @app.route('//convert/cancel/', methods=['POST']) @app.route('/convert/cancel/', subdomain='', methods=['POST']) @app.route('//convert/cancel/', subdomain='', methods=['POST']) def cancel_conversion(job_id, lang_code=None, filetype=None): """Cancel a pending or running conversion.""" import json as _json from local_jobs import _kill_job_procs db_path = os.path.join(dir_path, "storage", "sqlite.db") try: _kill_job_procs(job_id) except Exception: pass cancelled_result = _json.dumps({ 'error': True, 'message': 'Cancelled by user', 'results': [] }) with sqlite3.connect(db_path) as c: c.row_factory = sqlite3.Row c.execute( "UPDATE local_jobs SET status='failed', result_json=?, " "completed_at=datetime('now') WHERE id=? AND status IN ('queued','started')", (cancelled_result, job_id) ) conv = c.execute( "SELECT id FROM conversions WHERE job_id=? ORDER BY id DESC LIMIT 1", (job_id,) ).fetchone() if conv: c.execute( "UPDATE conversions SET status='cancelled', " "error_message='Cancelled by user' WHERE id=?", (conv['id'],) ) c.commit() flash("Conversion cancelled.", "info") return redirect(url_for('index')) @app.route('/convert/delete/', methods=['GET']) @app.route('//convert/delete/', methods=['GET']) @app.route('/convert/delete/', subdomain='', methods=['GET']) @app.route('//convert/delete/', subdomain='', methods=['GET']) def delete_conversion_get(job_id, lang_code=None): """GET /convert/delete/ — redirect to homepage. Old share/bookmark links that land on the delete URL should not 404; redirect to the homepage so the visitor lands somewhere useful.""" return redirect(url_for('index')) @app.route('/convert/delete/', methods=['POST']) @app.route('//convert/delete/', methods=['POST']) @app.route('/convert/delete/', subdomain='', methods=['POST']) @app.route('//convert/delete/', subdomain='', methods=['POST']) def delete_conversion(job_id, lang_code=None, filetype=None): """Delete output files and/or upload for a conversion.""" import json as _json, shutil as _sh from local_jobs import _kill_job_procs from helpers import admin_delete_file_upload del_output = request.form.get('delete_output') == 'on' del_upload = request.form.get('delete_upload') == 'on' if not del_output and not del_upload: flash("Please select at least one item to delete.", "warning") return redirect(request.referrer or url_for('index')) db_path = os.path.join(dir_path, "storage", "sqlite.db") try: _kill_job_procs(job_id) except Exception: pass conv = None with sqlite3.connect(db_path) as c: c.row_factory = sqlite3.Row conv = c.execute( "SELECT id, upload_id, output_path, file_name, user_id FROM conversions " "WHERE job_id=? ORDER BY id DESC LIMIT 1", (job_id,) ).fetchone() if del_output and conv: _upload_dir = app.config.get('UPLOAD_DIR', os.path.join(dir_path, 'storage', 'uploads')) output_path_raw = conv['output_path'] or '' for fpath in output_path_raw.split('\n'): fpath = fpath.strip() if not fpath: continue full = fpath if os.path.isabs(fpath) else os.path.join(_upload_dir, fpath) try: if os.path.isfile(full): os.remove(full) parent = os.path.dirname(full) if os.path.isdir(parent) and not os.listdir(parent): _sh.rmtree(parent, ignore_errors=True) except Exception: pass if del_upload and conv and conv['upload_id']: try: admin_delete_file_upload(conv['upload_id'], also_delete_conversions=False) except Exception: pass full_delete = del_output and del_upload if conv: try: with _db() as c: # Always hard-delete the conversion row and its upload record _upload_id = conv['upload_id'] if _upload_id: c.execute("DELETE FROM file_uploads WHERE id=?", (_upload_id,)) c.execute("DELETE FROM conversions WHERE id=?", (conv['id'],)) c.commit() except Exception: app.logger.exception("delete_conversion: conversions delete failed") raise try: with sqlite3.connect(db_path) as c: if full_delete: deleted_result = _json.dumps({ 'error': True, 'message': 'Deleted by user', 'results': [] }) c.execute( "UPDATE local_jobs SET status='failed', result_json=?, " "completed_at=datetime('now') WHERE id=?", (deleted_result, job_id) ) elif del_output: existing = c.execute( "SELECT result_json FROM local_jobs WHERE id=?", (job_id,) ).fetchone() if existing and existing[0]: try: rj = _json.loads(existing[0]) rj['results'] = [] rj['output_deleted'] = True except Exception: rj = {'error': False, 'results': [], 'output_deleted': True} else: rj = {'error': False, 'results': [], 'output_deleted': True} c.execute( "UPDATE local_jobs SET result_json=? WHERE id=?", (_json.dumps(rj), job_id) ) elif del_upload and not del_output: _existing = c.execute( "SELECT result_json FROM local_jobs WHERE id=?", (job_id,) ).fetchone() if _existing and _existing[0]: try: _rj = _json.loads(_existing[0]) _rj['upload_deleted'] = True except Exception: _rj = {'error': False, 'upload_deleted': True} else: _rj = {'error': False, 'upload_deleted': True} c.execute( "UPDATE local_jobs SET result_json=? WHERE id=?", (_json.dumps(_rj), job_id) ) c.commit() except Exception: app.logger.exception("delete_conversion: local_jobs update failed") # Revoke shared links and access grants on any delete path — not just full delete. # This ensures that once a conversion is marked deleted (output or upload gone), # any shared URL for this job immediately stops working. try: with _db() as c: c.execute("DELETE FROM shared_links WHERE job_id=?", (job_id,)) c.execute("DELETE FROM job_access_grants WHERE job_id=?", (job_id,)) c.commit() except Exception: app.logger.exception("delete_conversion: shared_links cleanup failed") try: _user = get_current_user() _conv_name = (conv['file_name'] if conv else None) or str(job_id) _actor_id = _user['id'] if _user else None _actor_name = _user.get('username', '') if _user else '' _owner_id = conv['user_id'] if conv else _actor_id log_deletion('conversion', entity_id=job_id, entity_name=_conv_name, actor_user_id=_actor_id, actor_username=_actor_name, actor_role='user', extra_meta={'job_id': job_id, 'del_output': del_output, 'del_upload': del_upload}, entity_owner_user_id=_owner_id) except Exception: app.logger.exception("delete_conversion: log_deletion failed") if full_delete: msg = "Converted output files and original upload have been permanently deleted." elif del_output: msg = "Converted output files have been deleted. Your original upload is still available." else: msg = "Original uploaded file has been deleted." flash(msg, "success") if full_delete: return redirect(url_for('index')) back_url = request.referrer if not back_url: try: back_url = url_for('output_route', job_id=job_id) except Exception: back_url = url_for('index') return redirect(back_url) @app.route('/api/job//notify-email', methods=['POST']) def api_job_notify_email(job_id): import re as _re ss = get_site_settings() if ss.get('job_notify_email_enabled', '0') != '1': return jsonify({'ok': False, 'error': 'Email notifications are disabled.'}), 403 data = request.get_json(silent=True) or {} email = (data.get('email') or '').strip().lower() if not email or not _re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', email): return jsonify({'ok': False, 'error': 'Please enter a valid email address.'}), 400 user = get_current_user() if user: plan = get_plan_by_id(user.get('plan_id')) if not plan or not int(plan.get('job_notify_email', 0)): return jsonify({'ok': False, 'error': 'Your plan does not include email notifications.'}), 403 else: free_plan = get_plan_by_id(1) if not free_plan or not int(free_plan.get('job_notify_email', 0)): return jsonify({'ok': False, 'error': 'Email notifications require a paid plan. Please sign in.'}), 403 db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: existing = conn.execute( "SELECT id FROM job_notify_emails WHERE job_id=? AND email=?", (job_id, email) ).fetchone() if existing: return jsonify({'ok': True, 'msg': 'Already subscribed.'}) conn.execute( "INSERT INTO job_notify_emails (job_id, email) VALUES (?,?)", (job_id, email) ) conn.commit() return jsonify({'ok': True, 'msg': 'You will be notified when conversion is done.'}) # ---- Push notifications ---- _job_push_locks = {} _job_push_locks_mu = __import__('threading').Lock() def _trigger_job_push_notifications(job_id, is_failed=False): with _job_push_locks_mu: if job_id in _job_push_locks: return _job_push_locks[job_id] = True try: ss = get_site_settings() if ss.get('job_notify_push_enabled', '0') != '1': return db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: conn.row_factory = sqlite3.Row # Job-specific subscriptions (per-output-page click) job_rows = conn.execute( "SELECT id, endpoint, p256dh, auth FROM job_notify_push WHERE job_id=? AND sent=0", (job_id,) ).fetchall() # Global subscriptions — look up via the job's owner (user or anon) conv_row = conn.execute( "SELECT user_id, anon_id FROM conversions WHERE job_id=? LIMIT 1", (job_id,) ).fetchone() global_rows = [] if conv_row: uid = conv_row['user_id'] aid = conv_row['anon_id'] or '' if uid: global_rows = conn.execute( "SELECT id, endpoint, p256dh, auth FROM user_push_subscriptions WHERE user_id=? AND active=1", (uid,) ).fetchall() elif aid: global_rows = conn.execute( "SELECT id, endpoint, p256dh, auth FROM user_push_subscriptions WHERE anon_id=? AND active=1", (aid,) ).fetchall() if not job_rows and not global_rows: return # Claim job-specific rows so they don't get re-sent if job_rows: claimed_ids = [r['id'] for r in job_rows] conn.execute( "UPDATE job_notify_push SET sent=2 WHERE id IN ({})".format( ','.join('?' * len(claimed_ids)) ), claimed_ids ) conn.commit() site_name = ss.get('smtp_from_name', 'OnlineConvert') or 'OnlineConvert' if is_failed: push_title = f"{site_name} – Conversion Failed" push_body = "Your conversion encountered an error. Please try again." else: push_title = f"{site_name} – Conversion Complete" push_body = "Your files are ready to download!" _site_url = ss.get('site_url', '').strip().rstrip('/') if not _site_url: _wu = website_url.rstrip('/') _site_url = _wu if _wu.startswith('http') else f'https://{_wu}' push_url = f"{_site_url}/output/{job_id}" sent_endpoints = set() # Send job-specific for row in job_rows: ok = send_push_notification( row['endpoint'], row['p256dh'], row['auth'], title=push_title, body=push_body, url=push_url ) conn.execute("UPDATE job_notify_push SET sent=? WHERE id=?", (1 if ok is True else 0, row['id'])) sent_endpoints.add(row['endpoint']) # Send global (deduplicate endpoints already handled above) for row in global_rows: if row['endpoint'] in sent_endpoints: continue ok = send_push_notification( row['endpoint'], row['p256dh'], row['auth'], title=push_title, body=push_body, url=push_url ) if ok is not True: # Stale, expired, or failed — remove the subscription record conn.execute( "DELETE FROM user_push_subscriptions WHERE id=?", (row['id'],) ) sent_endpoints.add(row['endpoint']) conn.commit() except Exception as e: app.logger.error(f"Job push notify error for {job_id}: {e}") finally: with _job_push_locks_mu: _job_push_locks.pop(job_id, None) @app.route('/api/vapid-public-key', methods=['GET']) def api_vapid_public_key(): ss = get_site_settings() if ss.get('job_notify_push_enabled', '0') != '1': return jsonify({'ok': False, 'error': 'Push notifications are disabled.'}), 403 try: _, pub_key = get_vapid_keys() return jsonify({'ok': True, 'publicKey': pub_key}) except ImportError as e: return jsonify({'ok': False, 'error': str(e)}), 503 except Exception as e: return jsonify({'ok': False, 'error': f'Failed to get VAPID key: {str(e)}'}), 503 @app.route('/api/job//notify-push', methods=['POST']) def api_job_notify_push(job_id): ss = get_site_settings() if ss.get('job_notify_push_enabled', '0') != '1': return jsonify({'ok': False, 'error': 'Push notifications are disabled.'}), 403 data = request.get_json(silent=True) or {} endpoint = (data.get('endpoint') or '').strip() p256dh = (data.get('p256dh') or '').strip() auth = (data.get('auth') or '').strip() if not endpoint or not p256dh or not auth: return jsonify({'ok': False, 'error': 'Invalid subscription data.'}), 400 db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: existing = conn.execute( "SELECT id FROM job_notify_push WHERE job_id=? AND endpoint=?", (job_id, endpoint) ).fetchone() if existing: return jsonify({'ok': True, 'msg': 'Already subscribed.'}) conn.execute( "INSERT INTO job_notify_push (job_id, endpoint, p256dh, auth) VALUES (?,?,?,?)", (job_id, endpoint, p256dh, auth) ) conn.commit() return jsonify({'ok': True, 'msg': 'You will receive a push notification when done.'}) @app.route('/api/push/subscribe-global', methods=['POST']) def api_push_subscribe_global(): ss = get_site_settings() if ss.get('job_notify_push_enabled', '0') != '1': return jsonify({'ok': False, 'error': 'Push notifications are disabled.'}), 403 data = request.get_json(silent=True) or {} endpoint = (data.get('endpoint') or '').strip() p256dh = (data.get('p256dh') or '').strip() auth = (data.get('auth') or '').strip() if not endpoint or not p256dh or not auth: return jsonify({'ok': False, 'error': 'Invalid subscription data.'}), 400 user = get_current_user() user_id = user['id'] if user else None anon_id = '' if user_id else (request.cookies.get('oc_push_anon') or '') if not user_id and not anon_id: return jsonify({'ok': False, 'error': 'No identity available.'}), 400 try: db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: conn.execute(""" INSERT INTO user_push_subscriptions (user_id, anon_id, endpoint, p256dh, auth, active, updated_at) VALUES (?,?,?,?,?,1,datetime('now')) ON CONFLICT(endpoint) DO UPDATE SET user_id=excluded.user_id, anon_id=excluded.anon_id, p256dh=excluded.p256dh, auth=excluded.auth, active=1, updated_at=datetime('now') """, (user_id, anon_id, endpoint, p256dh, auth)) conn.commit() return jsonify({'ok': True}) except Exception as e: return jsonify({'ok': False, 'error': str(e)}), 500 @app.route('/api/push/global-status', methods=['GET']) def api_push_global_status(): ss = get_site_settings() if ss.get('job_notify_push_enabled', '0') != '1': return jsonify({'ok': True, 'subscribed': False, 'push_enabled': False}) user = get_current_user() user_id = user['id'] if user else None anon_id = '' if user_id else (request.cookies.get('oc_push_anon') or '') try: db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: if user_id: row = conn.execute( "SELECT id FROM user_push_subscriptions WHERE user_id=? AND active=1 LIMIT 1", (user_id,) ).fetchone() elif anon_id: row = conn.execute( "SELECT id FROM user_push_subscriptions WHERE anon_id=? AND active=1 LIMIT 1", (anon_id,) ).fetchone() else: row = None return jsonify({'ok': True, 'subscribed': row is not None, 'push_enabled': True}) except Exception: return jsonify({'ok': True, 'subscribed': False, 'push_enabled': True}) @app.route('/api/push/unsubscribe-global', methods=['POST']) def api_push_unsubscribe_global(): """Delete push subscription records for the current user/anon/endpoint.""" user = get_current_user() user_id = user['id'] if user else None anon_id = '' if user_id else (request.cookies.get('oc_push_anon') or '') data = request.get_json(silent=True) or {} endpoint = (data.get('endpoint') or '').strip() try: db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as conn: if endpoint: # Delete the specific browser endpoint record conn.execute( "DELETE FROM user_push_subscriptions WHERE endpoint=?", (endpoint,) ) elif user_id: conn.execute( "DELETE FROM user_push_subscriptions WHERE user_id=?", (user_id,) ) elif anon_id: conn.execute( "DELETE FROM user_push_subscriptions WHERE anon_id=?", (anon_id,) ) conn.commit() return jsonify({'ok': True}) except Exception as e: return jsonify({'ok': False, 'error': str(e)}), 500 def render_converter_page(filetype, target_format, source_format=None): """Render the actual converter template""" app.logger.info(f"Rendering converter: {filetype} -> {target_format}, source: {source_format}") # Validate target format valid_formats = [f.lower() for f in available_filetypes[filetype]['ext']] if target_format.lower() not in valid_formats: app.logger.info(f"Target format {target_format} not valid for {filetype}") abort(404) # Validate source format if provided if source_format: allowed_formats = [f.lower() for f in available_filetypes[filetype]['allowed']] if source_format.lower() not in allowed_formats: app.logger.info(f"Source format {source_format} not allowed for {filetype}") abort(404) # Render appropriate template client_side_tools = available_filetypes[filetype].get('client_side_tools', []) if target_format.lower() in client_side_tools: return render_template('converter-pdf.html', filetypes=available_filetypes, filetype=filetype, fileformat=target_format, source_format=source_format, tool_meta=PDF_TOOL_META, tu=target_format) import copy as _copy _raw_opts = available_filetypes[filetype]['options'] _opts = _copy.deepcopy(_raw_opts) _lang = getattr(g, 'lang_code', 'en') or 'en' _page_ids = [] if source_format and target_format: # Admin stores pair pages as converter:{type}:{src}:{tgt} (colon-separated) _page_ids.append(f"converter:{filetype}:{source_format.lower()}:{target_format.lower()}") # Legacy / alternate hyphen format kept for backwards compatibility _page_ids.append(f"converter:{filetype}:{source_format.lower()}-to-{target_format.lower()}") if target_format: _page_ids.append(f"converter:{filetype}:{target_format.lower()}") _page_ids.append(f"converter:{filetype}") def _cms_opt(prop_key, field): for _pid in _page_ids: _v = get_page_content(_pid, f"option:{prop_key}:{field}", _lang) if _v: return _v _v = get_page_content(_pid, f"option:{prop_key}:{field}", 'en') if _v: return _v return '' for _prop_key, _prop in (_opts.get('properties') or {}).items(): _t = _cms_opt(_prop_key, 'title') if _t: _prop['title'] = _t _d = _cms_opt(_prop_key, 'description') if _d: _prop['description'] = _d options_json = json.dumps(_opts, indent=2) _user = get_current_user() _plan = get_plan_by_id(_user['plan_id']) if _user else get_plan_by_id(1) _max_files = int(_plan['max_files_per_day']) if _plan and _plan.get('max_files_per_day') and int(_plan['max_files_per_day']) > 0 else 10 _raw_batch = int(_plan['max_batch_files']) if _plan and _plan.get('max_batch_files') is not None else None _max_batch = 9999 if _raw_batch == 0 else (_raw_batch if _raw_batch and _raw_batch > 0 else 20) _ss = get_site_settings() _server_max_mb = int(_ss.get('server_max_upload_mb', 500)) _plan_max_mb = int(_plan['max_file_size_mb']) if _plan and _plan.get('max_file_size_mb') and int(_plan['max_file_size_mb']) > 0 else None _effective_max_mb = min(_server_max_mb, _plan_max_mb) if _plan_max_mb is not None else _server_max_mb import hmac as _hmac, hashlib as _hashlib _ctx_ft = filetype or '' _ctx_sf = source_format or '' _ctx_str = f"{_ctx_ft}:{_ctx_sf}" _ctx_sig = _hmac.new( app.secret_key.encode() if isinstance(app.secret_key, str) else app.secret_key, _ctx_str.encode(), _hashlib.sha256 ).hexdigest() _url_prefetch_token = f"{_ctx_str}:{_ctx_sig}" return render_template('converter.html', filetypes=available_filetypes, filetype=filetype, fileformat=target_format, source_format=source_format, options=options_json, tu=target_format, plan_max_files=_max_files, plan_max_batch_files=_max_batch, plan_max_size_bytes=_effective_max_mb * 1024 * 1024, plan_max_size_mb=_effective_max_mb, site_settings=_ss, url_prefetch_token=_url_prefetch_token) def render_filetype_list(filetype): """Render the list page for a file type""" app.logger.info(f"Rendering list for filetype: {filetype}") if filetype == "hash": return render_template('list2.html', filetypes=available_hashtypes, filetype=filetype, fileformat=None) if filetype not in available_filetypes: abort(404) extra = {} if filetype == 'pdf': extra['tool_meta'] = PDF_TOOL_META return render_template('list.html', filetypes=available_filetypes, filetype=filetype, fileformat=None, **extra) def find_filetype_for_formats(format1, format2): """Find which file type handles conversion between these formats""" format1_lower = format1.lower() format2_lower = format2.lower() for ft, config in available_filetypes.items(): allowed = [f.lower() for f in config.get('allowed', [])] extensions = [f.lower() for f in config.get('ext', [])] if format1_lower in allowed and format2_lower in extensions: return ft return None def _resolve_urls(urls, upload_dir, _errors=None): """ Download any raw http(s):// URLs to local files and replace them with a JSON path-entry so all converters receive only local paths. Handles: - Plain URLs typed/pasted by the user - Google Drive (https://drive.google.com/uc?export=download&id=...) - Dropbox direct links (https://dl.dropboxusercontent.com/...) - OneDrive @microsoft.graph.downloadUrl links - Any other direct-download URL Plain JSON path-entries (already uploaded files) are returned unchanged. Optional _errors list: if provided, human-readable failure reasons are appended to it for each URL that could not be downloaded. """ _HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Accept": "*/*", } def _filename_from_response(resp, fallback_url): """Extract filename from Content-Disposition or URL path.""" cd = resp.headers.get("Content-Disposition", "") if cd: # filename="foo.jpg" or filename*=UTF-8''foo.jpg import re as _re2 m = _re2.search(r'filename\*?=["\']?(?:UTF-8\'\')?([^"\';\r\n]+)', cd, _re2.IGNORECASE) if m: name = m.group(1).strip().strip('"\'') name = secure_filename(name) if name: return name parsed_path = _urlparse(fallback_url).path name = parsed_path.rstrip('/').split('/')[-1] or '' name = secure_filename(name) return name or (uuid.uuid4().hex + '.bin') def _download(url): """Download url → local file, return relative path string or None.""" url = url.strip() # ── Google Drive large-file bypass ────────────────────────────── # drive.google.com/uc?export=download&id=... # drive.google.com/file/d//view → normalise to uc?export=download import re as _re2 gd_file = _re2.search(r'drive\.google\.com/file/d/([^/?#]+)', url) if gd_file: url = f"https://drive.google.com/uc?export=download&id={gd_file.group(1)}" sess = _requests.Session() sess.headers.update(_HEADERS) resp = sess.get(url, timeout=30, stream=True, allow_redirects=True) resp.raise_for_status() # Google Drive warning page for large files ct = resp.headers.get("Content-Type", "") if "drive.google.com" in url and "text/html" in ct: # Extract confirm token from the warning page confirm_m = _re2.search(r'confirm=([0-9A-Za-z_\-]+)', resp.text) if confirm_m: url = url + "&confirm=" + confirm_m.group(1) else: url = url + "&confirm=t" resp = sess.get(url, timeout=60, stream=True, allow_redirects=True) resp.raise_for_status() remote_name = _filename_from_response(resp, url) folder = uuid.uuid4().hex folder_path = os.path.join(upload_dir, folder) os.makedirs(folder_path, exist_ok=True) save_path = os.path.join(folder_path, remote_name) with open(save_path, 'wb') as f: for chunk in resp.iter_content(chunk_size=65536): if chunk: f.write(chunk) return f"{folder}/{remote_name}" resolved = [] for entry in urls: if not isinstance(entry, str): resolved.append(entry) continue entry = entry.strip() # Parse JSON entries — two distinct shapes come in: # A) Already-uploaded local file: {"path": "folder/file.ext", ...} # B) URL-tab submission from JS: {"name":"f.jpg","url":"https://...","type":"url"} if entry.startswith('{'): try: obj = json.loads(entry) if isinstance(obj, dict): # Shape B — URL file added via the URL input tab if obj.get('type') == 'url' and obj.get('url', '').lower().startswith(('http://', 'https://')): try: local_rel = _download(obj['url']) resolved.append(json.dumps({"path": local_rel})) except Exception as _e: app.logger.warning(f"_resolve_urls: failed to download {obj['url']!r}: {_e}") if _errors is not None: _errors.append(str(_e)) continue # Shape A — local path entry, pass through unchanged if 'path' in obj: resolved.append(entry) continue except (json.JSONDecodeError, TypeError): pass # Malformed JSON — skip continue # Raw remote URL (plain string) — covers plain URLs, Google Drive, Dropbox, OneDrive if entry.lower().startswith(('http://', 'https://')): try: local_rel = _download(entry) resolved.append(json.dumps({"path": local_rel})) except Exception as _e: app.logger.warning(f"_resolve_urls: failed to download {entry!r}: {_e}") if _errors is not None: _errors.append(str(_e)) continue # Anything else (already a local relative path string) — pass through resolved.append(entry) return resolved def _get_job_queue_name(): """Return 'high', 'medium', or 'low' based on the current user's plan.""" try: _uid = session.get('user_id') if not _uid: return 'low' _user = get_current_user() if not _user: return 'low' _plan = get_plan_by_id(_user['plan_id']) if not _plan: return 'low' q_name = (_plan.get('queue') or 'low').strip().lower() return q_name if q_name in ('high', 'medium', 'low') else 'low' except Exception: return 'low' def _get_plan_timeout_seconds(): """Return the plan's max_processing_seconds (0 = unlimited).""" try: _uid = session.get('user_id') _user = get_current_user() if _uid else None _plan = get_plan_by_id(_user['plan_id']) if _user else get_plan_by_id(1) if not _plan: return 0 return int(_plan.get('max_processing_seconds') or 0) except Exception: return 0 def process_conversion_job(filetype, fileformat, urls, options_dict): """Create and enqueue a conversion job""" # Download any raw remote URLs to local files before handing off to the converter urls = _resolve_urls(urls, app.config['UPLOAD_DIR']) _conv_args = (urls, fileformat, options_dict, { "UPLOAD_DIR": app.config['UPLOAD_DIR'], "BUCKET": app.config['BUCKET'], "LOCAL": app.config['LOCAL'] }) # Determine priority queue and timeout based on the user's plan _queue_name = _get_job_queue_name() _plan_timeout = _get_plan_timeout_seconds() job = _local_queue.enqueue_call( func=convert_list[filetype].convert, args=_conv_args, result_ttl=5000, timeout=_plan_timeout, queue_name=_queue_name ) # Store in database try: _user_id = session.get('user_id') _upload_id = request.form.get('upload_id') def _parse_url_entry(u): try: d = json.loads(u) if isinstance(d, dict): return d.get('path', u), d.get('upload_id') except Exception: pass return u, None _raw_url = urls[0] if urls else '' _parsed_path, _parsed_uid = _parse_url_entry(_raw_url) if not _upload_id and _parsed_uid: _upload_id = _parsed_uid _file_name = _parsed_path.split('/')[-1] if _parsed_path else '' _input_fmt = (_file_name.rsplit('.', 1)[-1].lower() if _file_name and '.' in _file_name else filetype) _conv_ip = get_client_ip() _conv_tool = 'hash' if filetype == 'hash' else ('pdf' if filetype == 'pdf' else 'converter') _options_json = json.dumps({"filetype": filetype, "options": options_dict}) _anon_id = '' if _user_id else (request.cookies.get('oc_push_anon') or '') with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as _cc: _cc.execute( "INSERT INTO conversions (user_id, upload_id, input_format, output_format, job_id, status, file_name, ip_address, tool_type, queue, options_json, anon_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (_user_id, _upload_id, _input_fmt, fileformat, job.id, 'pending', _file_name, _conv_ip, _conv_tool, _queue_name, _options_json, _anon_id) ) _cc.commit() except Exception: pass return job # ===== LEGACY HANDLER FOR CONVERT_TO ROUTE ===== def _convert_to_handler(filetype, subpath): """Legacy handler for convert_to route""" # Handle language prefix first parts = subpath.split('/', 1) lang_code = None remaining = subpath if parts[0] in supported_languages: lang_code = parts[0] remaining = parts[1] if len(parts) > 1 else "" app.logger.info(f"Detected language: {lang_code}, remaining: {remaining}") # Handle output paths with language if remaining.startswith('output/'): job_id = remaining.replace('output/', '').split('/')[0] app.logger.info(f"Output path with job: {job_id}, lang: {lang_code}") if lang_code: # Redirect to language-specific subdomain output route return redirect(url_for('subdomain_output_lang', filetype=filetype, lang_code=lang_code, job_id=job_id, _external=True)) else: # Redirect to regular subdomain output route return redirect(url_for('subdomain_output', filetype=filetype, job_id=job_id, _external=True)) # Handle convert-to paths if remaining.startswith('convert-to-'): fileformat = remaining.replace('convert-to-', '') if lang_code: g.lang_code = lang_code return render_converter_page(filetype, fileformat) # Handle convert/x-to-y paths if remaining.startswith('convert/') and '-to-' in remaining: path_part = remaining.replace('convert/', '') if '-to-' in path_part: source, target = path_part.split('-to-', 1) config = available_filetypes.get(filetype) if config: allowed = [f.lower() for f in config.get('allowed', [])] extensions = [f.lower() for f in config.get('ext', [])] if source.lower() in allowed and target.lower() in extensions: if lang_code: g.lang_code = lang_code return render_converter_page(filetype, target, source) # If no subpath, show list page if not remaining: if lang_code: g.lang_code = lang_code return render_filetype_list(filetype) app.logger.info(f"No match found for {filetype}/{subpath}, aborting 404") abort(404) # ===== INDEX ROUTE (MUST BE FIRST) ===== @app.route("/", subdomain="") @app.route("/", subdomain="") @app.route("//", subdomain="") @app.route("/") @app.route("//") @app.route("/") def index(subdomain=None, lang_code=None): """Main index page and subdomain homepages""" if subdomain is None: subdomain = _filetype_from_host() if subdomain and lang_code and lang_code not in supported_languages: return _convert_to_handler(subdomain.lower(), lang_code) if subdomain: filetype = subdomain.lower() # If subdomains are disabled but we're on a subdomain, redirect to main domain if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{lang_code}" if lang_code and lang_code != "en" else "" return redirect(f"{request.scheme}://{server_name}{lang_prefix}/converter/{filetype}") if filetype == "hash": return render_template('list2.html', filetypes=available_hashtypes, filetype=filetype, fileformat=None) if filetype not in available_filetypes: abort(404) return render_filetype_list(filetype) # Main domain index page return render_template('index.html', filetypes=available_filetypes, hashtypes=available_hashtypes) # ===== LEGACY CONVERTER ROUTES (keep for backward compatibility) ===== @app.route('/converter/', methods=['GET']) @app.route('/CONVERTER/', methods=['GET']) @app.route("//converter/", methods=['GET']) @app.route('/converter//', methods=['GET', 'POST']) @app.route('/CONVERTER//', methods=['GET', 'POST']) @app.route("//converter//", methods=['GET', 'POST']) def converter(filetype=None, fileformat=None, first=None, second=None): """Legacy converter routes - redirect to new structure based on subdomain setting""" # If this is a list page (no fileformat) if fileformat == None and first == None and second == None: filetype = filetype.lower() # If subdomains are enabled, redirect to subdomain homepage if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype}.{server_name}{lang_prefix}/") return render_filetype_list(filetype) # PDF tools live at /do/ only — /converter/pdf/ is not a valid URL if filetype.lower() == 'pdf': abort(404) # Handle GET request for converter page if request.method == "GET": filetype2 = filetype.lower() _ft_cfg = available_filetypes.get(filetype2) or available_hashtypes.get(filetype2, {}) filees = _ft_cfg.get('ext', []) ku = fileformat.lower() if fileformat and fileformat.lower() in map(str.lower, filees) else (fileformat or '').lower() # If subdomains are enabled, redirect to subdomain if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype2}.{server_name}{lang_prefix}/convert-to-{ku}") _all_filetypes = {**available_filetypes, **available_hashtypes} return render_template('converter.html', filetypes=_all_filetypes, filetype=filetype2, fileformat=ku, options=json.dumps(_ft_cfg.get('options', {}), sort_keys=False, indent=2), tu=fileformat) # Handle POST request (file upload) else: urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url("/converter/{}/{}".format(filetype, fileformat))) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url("/converter/{}/{}".format(filetype, fileformat))) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" _host = request.host.split(':')[0] _parts = _host.split('.') _base = '.'.join(_parts[-2:]) if len(_parts) > 2 else _host if _is_subdomains_enabled(): return redirect(f"{request.scheme}://{filetype}.{_base}{lang_prefix}/output/{job.id}") return redirect(f"{lang_prefix}/output/{job.id}") # ===== NEW STRUCTURED ROUTES ===== # Main domain routes (when subdomains OFF) @app.route('/converter2/', methods=['GET']) def converter_list(filetype): """onlineconvert.cc/converter/image""" filetype = filetype.lower() if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype}.{server_name}{lang_prefix}/") return render_filetype_list(filetype) @app.route('/converter2//', methods=['GET', 'POST']) def converter_page(filetype, fileformat): """onlineconvert.cc/converter/image/jpg""" filetype = filetype.lower() fileformat = fileformat.lower() if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype}.{server_name}{lang_prefix}/convert-to-{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('converter_page', filetype=filetype, fileformat=fileformat)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('converter_page', filetype=filetype, fileformat=fileformat)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" return redirect(f"{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) # ===== DEVICE CONVERTER ROUTE ===== @app.route('//convert-for-', subdomain='', methods=['GET', 'POST']) @app.route('/convert-for-', subdomain='', methods=['GET', 'POST']) def subdomain_convert_for(filetype, fileformat): """Handle device converter URLs: device.onlineconvert.cc/convert-for-psp""" app.logger.info(f"===== SUBDOMAIN_CONVERT_FOR HIT =====") app.logger.info(f"filetype: {filetype}") app.logger.info(f"fileformat: {fileformat}") filetype = filetype.lower() fileformat = fileformat.lower() # Only allow this pattern for device subdomains if filetype not in ["device", "webservice"]: app.logger.info("Only device or webservice subdomains can use /convert-for- pattern. Returning 404.") abort(404) if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{server_name}{lang_prefix}/converter/{filetype}/{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('subdomain_convert_for', filetype=filetype, fileformat=fileformat, _external=True)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('subdomain_convert_for', filetype=filetype, fileformat=fileformat, _external=True)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" _host = request.host.split(':')[0] return redirect(f"{request.scheme}://{_host}{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) @app.route('//do/', methods=['GET', 'POST']) @app.route('/do/', methods=['GET', 'POST']) def pdf_tool_main(fileformat, lang_code=None): """Main-domain PDF tool route — works when subdomains are OFF. e.g. domain.com/do/merge-pdf """ if lang_code and lang_code in supported_languages: g.lang_code = lang_code fileformat = fileformat.lower() if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_pfx = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://pdf.{server_name}{lang_pfx}/do/{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('pdf_tool_main', fileformat=fileformat)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('pdf_tool_main', fileformat=fileformat)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job('pdf', fileformat, urls, options) lang_pfx = f"/{getattr(g, 'lang_code', 'en')}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{lang_pfx}/output/{job.id}") return render_converter_page('pdf', fileformat) @app.route('//do/', subdomain='', methods=['GET', 'POST']) @app.route('/do/', subdomain='', methods=['GET', 'POST']) def subdomain_convert_for_pdf(filetype, fileformat): """Handle device converter URLs: device.onlineconvert.cc/convert-for-psp""" app.logger.info(f"===== SUBDOMAIN_CONVERT_FOR HIT =====") app.logger.info(f"filetype: {filetype}") app.logger.info(f"fileformat: {fileformat}") filetype = filetype.lower() fileformat = fileformat.lower() # Only allow this pattern for device subdomains if filetype != "pdf": app.logger.info(f"Non-device subdomain cannot use /convert-for- pattern. Returning 404.") abort(404) if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{server_name}{lang_prefix}/converter/{filetype}/{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('subdomain_convert_for', filetype=filetype, fileformat=fileformat, _external=True)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('subdomain_convert_for', filetype=filetype, fileformat=fileformat, _external=True)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" _host = request.host.split(':')[0] return redirect(f"{request.scheme}://{_host}{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) # ===== COMPRESSOR ROUTES ===== @app.route('//compress-', subdomain='', methods=['GET', 'POST']) @app.route('/compress-', subdomain='', methods=['GET', 'POST']) def subdomain_compress(filetype, fileformat): """Handle compressor URLs: image-compressor.onlineconvert.cc/compress-jpg""" app.logger.info(f"===== SUBDOMAIN_COMPRESS HIT =====") app.logger.info(f"filetype: {filetype}") app.logger.info(f"fileformat: {fileformat}") filetype = filetype.lower() fileformat = fileformat.lower() # Only allow this pattern for compressor subdomains if filetype not in ["image-compressor", "video-compressor"]: app.logger.info(f"Non-compressor subdomain cannot use /compress- pattern. Returning 404.") abort(404) if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{server_name}{lang_prefix}/converter/{filetype}/{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('subdomain_compress', filetype=filetype, fileformat=fileformat, _external=True)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('subdomain_compress', filetype=filetype, fileformat=fileformat, _external=True)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" _host = request.host.split(':')[0] return redirect(f"{request.scheme}://{_host}{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) # ===== HASH GENERATOR SUBDOMAIN ROUTE ===== @app.route('//converter/-to-', methods=['GET', 'POST']) @app.route('/converter/-to-', methods=['GET', 'POST']) def converter_slug(first, second): """somedomain.com/converter/eps-to-bmp (both-ext slug, no subdomain)""" first = first.lower() second = second.lower() filetype = find_filetype_for_formats(first, second) if not filetype: abort(404) if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) current_host = request.host.split(':')[0].lower() _sn_clean = server_name.split(':')[0].lower() if current_host == _sn_clean or current_host.endswith('.' + _sn_clean): lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype}.{server_name}{lang_prefix}/convert/{first}-to-{second}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('converter_slug', first=first, second=second)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('converter_slug', first=first, second=second)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, second, urls, options) # job is already recorded by process_conversion_job lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" return redirect(f"{lang_prefix}/output/{job.id}") return render_converter_page(filetype, second, first) @app.route('//convert-to-', methods=['GET', 'POST']) @app.route('/convert-to-', methods=['GET', 'POST']) def convert_to_fmt(fileformat, lang_code=None): """Non-subdomain route: /convert-to-zip → archive converter""" fileformat = fileformat.lower() filetype = None for ft, cfg in available_filetypes.items(): if ft in ('device', 'webservice'): continue if fileformat in [f.lower() for f in cfg.get('ext', [])]: filetype = ft break if not filetype: abort(404) if filetype == 'pdf': lang_prefix = f"/{getattr(g, 'lang_code', 'en')}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{lang_prefix}/do/{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('convert_to_fmt', fileformat=fileformat)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('convert_to_fmt', fileformat=fileformat)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" return redirect(f"{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) @app.route('//convert-for-', methods=['GET', 'POST']) @app.route('/convert-for-', methods=['GET', 'POST']) def convert_for_fmt(fileformat, lang_code=None): """Non-subdomain route: /convert-for-psp → device/webservice converter""" fileformat = fileformat.lower() filetype = None for ft in ('device', 'webservice'): cfg = available_filetypes.get(ft, {}) if fileformat in [f.lower() for f in cfg.get('ext', [])]: filetype = ft break if not filetype: abort(404) if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('convert_for_fmt', fileformat=fileformat)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('convert_for_fmt', fileformat=fileformat)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" return redirect(f"{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) @app.route('//convert/-to-', methods=['GET', 'POST']) @app.route('/convert/-to-', methods=['GET', 'POST']) def convert_direct(first, second): """onlineconvert.cc/convert/jpg-to-png""" first = first.lower() second = second.lower() filetype = find_filetype_for_formats(first, second) if not filetype: abort(404) if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) current_host = request.host.split(':')[0].lower() _sn_clean = server_name.split(':')[0].lower() # Only redirect when the request is actually on the configured domain if current_host == _sn_clean or current_host.endswith('.' + _sn_clean): lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype}.{server_name}{lang_prefix}/convert/{first}-to-{second}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('convert_direct', first=first, second=second)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('convert_direct', first=first, second=second)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, second, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" return redirect(f"{lang_prefix}/output/{job.id}") return render_converter_page(filetype, second, first) # ===== FORMAT-CONVERTER SLUG ROUTES ===== # /png-converter (short form — auto-detect filetype and redirect) # NOTE: no //-converter variant — would conflict with //-converter @app.route('/-converter', methods=['GET']) def short_format_converter(fileformat): """e.g. domain.com/png-converter → auto-detect filetype → redirect to /image/png-converter""" fileformat = fileformat.lower() _all_types = {**available_filetypes, **available_hashtypes} # find which filetype this format belongs to detected = None for ft, cfg in _all_types.items(): if fileformat in [f.lower() for f in cfg.get('ext', [])]: detected = ft break if not detected: abort(404) lang_pfx = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{lang_pfx}/{detected}/{fileformat}-converter", 301) # /image/jpg-converter (main domain, with explicit filetype) @app.route('///-converter', methods=['GET']) @app.route('//-converter', methods=['GET']) def format_converter_page(filetype, fileformat, lang_code=None): """Format info + selector page — e.g. domain.com/image/jpg-converter. No file upload here; user picks source/target and is redirected to /convert/src-to-tgt.""" if lang_code and lang_code in supported_languages: g.lang_code = lang_code filetype = filetype.lower() fileformat = fileformat.lower() # Validate filetype — prevents catching language codes or unknown paths _all_filetypes = {**available_filetypes, **available_hashtypes} if filetype not in _all_filetypes: abort(404) if _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_pfx = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{filetype}.{server_name}{lang_pfx}/{fileformat}-converter") # Validate format belongs to this filetype _ft_cfg = _all_filetypes.get(filetype, {}) valid_exts = [f.lower() for f in _ft_cfg.get('ext', [])] if valid_exts and fileformat not in valid_exts: abort(404) return render_template('convert-selector.html', filetypes=available_filetypes, filetype=filetype, fileformat=fileformat) # image.domain/jpg-converter (subdomain) @app.route('//-converter', subdomain='', methods=['GET']) @app.route('/-converter', subdomain='', methods=['GET']) def subdomain_format_converter(filetype, fileformat, lang_code=None): """e.g. image.onlineconvert.cc/jpg-converter — info/selector page, no upload.""" if lang_code and lang_code in supported_languages: g.lang_code = lang_code filetype = filetype.lower() fileformat = fileformat.lower() if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_pfx = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{server_name}{lang_pfx}/{filetype}/{fileformat}-converter") _ft_cfg = {**available_filetypes, **available_hashtypes}.get(filetype, {}) valid_exts = [f.lower() for f in _ft_cfg.get('ext', [])] if valid_exts and fileformat not in valid_exts: abort(404) return render_template('convert-selector.html', filetypes=available_filetypes, filetype=filetype, fileformat=fileformat) # Subdomain routes (when subdomains ON) @app.route('//convert-to-', subdomain='', methods=['GET', 'POST']) @app.route('/convert-to-', subdomain='', methods=['GET', 'POST']) def subdomain_convert_to(filetype, fileformat): """filetype.onlineconvert.cc/convert-to-jpg""" filetype = filetype.lower() fileformat = fileformat.lower() if filetype in ["device", "webservice"]: app.logger.info("Only device or webservice subdomains can use /convert-for- pattern. Returning 404.") abort(404) if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{server_name}{lang_prefix}/converter/{filetype}/{fileformat}") if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('subdomain_convert_to', filetype=filetype, fileformat=fileformat, _external=True)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('subdomain_convert_to', filetype=filetype, fileformat=fileformat, _external=True)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, fileformat, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" _host = request.host.split(':')[0] return redirect(f"{request.scheme}://{_host}{lang_prefix}/output/{job.id}") return render_converter_page(filetype, fileformat) @app.route('//convert/-to-', subdomain='', methods=['GET', 'POST']) @app.route('/convert/-to-', subdomain='', methods=['GET', 'POST']) def subdomain_convert_direct(filetype, first, second): """filetype.onlineconvert.cc/convert/bmp-to-jpg""" filetype = filetype.lower() first = first.lower() second = second.lower() if not _is_subdomains_enabled(): server_name = (app.config.get('SERVER_NAME') or website_url) lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" return redirect(f"{request.scheme}://{server_name}{lang_prefix}/convert/{first}-to-{second}") # CRITICAL VALIDATION: Ensure both formats belong to this filetype config = available_filetypes.get(filetype) if not config: abort(404) allowed_formats = [f.lower() for f in config.get('allowed', [])] target_formats = [f.lower() for f in config.get('ext', [])] if first not in allowed_formats or second not in target_formats: abort(404) if request.method == 'POST': urls = request.form.getlist('url[]') if len(urls) == 0: flash(_("Please upload a file to convert."), "error") return redirect(url_for('subdomain_convert_direct', filetype=filetype, first=first, second=second, _external=True)) _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(url_for('subdomain_convert_direct', filetype=filetype, first=first, second=second, _external=True)) options = json.loads(request.form.get('options', '{}')) job = process_conversion_job(filetype, second, urls, options) lang = getattr(g, 'lang_code', 'en') lang_prefix = f"/{lang}" if lang != "en" else "" _host = request.host.split(':')[0] return redirect(f"{request.scheme}://{_host}{lang_prefix}/output/{job.id}") return render_converter_page(filetype, second, first) # ===== SUBDOMAIN OUTPUT ROUTES (FIX FOR THE 500 ERROR) ===== @app.route('//output/', subdomain='', methods=['GET']) @app.route('/output/', subdomain='', methods=['GET']) def subdomain_output(filetype, job_id): """Handle output pages on subdomains: image.onlineconvert.cc/output/job_id""" app.logger.info(f"===== SUBDOMAIN OUTPUT HIT =====") app.logger.info(f"filetype: {filetype}, job_id: {job_id}") return output(job_id) # ===== HASH GENERATOR ROUTES ===== # Main domain hash routes (with /hash/ prefix) @app.route("//hash", methods=['GET', 'POST']) @app.route('/hash', methods=['GET', 'POST']) @app.route("//hash/-generator", methods=['GET', 'POST']) @app.route('/hash/-generator', methods=['GET', 'POST']) @app.route('/HASH/-generator', methods=['GET', 'POST']) # Hash subdomain routes (hash.onlineconvert.cc) @app.route('//-generator', subdomain='hash', methods=['GET', 'POST']) @app.route('/-generator', subdomain='hash', methods=['GET', 'POST']) # Also support hash-generators subdomain for backward compatibility @app.route('//-generator', subdomain='hash-generators', methods=['GET', 'POST']) @app.route('/-generator', subdomain='hash-generators', methods=['GET', 'POST']) def hash(hashformat=None, filetype='hash', first=None, second=None): app.logger.info(f"===== HASH FUNCTION HIT =====") app.logger.info(f"hashformat: {hashformat}") app.logger.info(f"filetype: {filetype}") app.logger.info(f"request.url: {request.url}") app.logger.info(f"subdomain: {_filetype_from_host()}") # If this is a list page (no hashformat) if hashformat is None and first is None and second is None: return render_template('list2.html', filetypes=available_hashtypes, filetype=filetype, fileformat=None) hashformat_input = hashformat.lower() filees = available_hashtypes[filetype]['ext'] if hashformat_input not in map(str.lower, filees): app.logger.info(f"Hash format {hashformat_input} not found") abort(404) hash_options = available_hashtypes[filetype]['options'] options_key = next((k for k in hash_options if k.lower() == hashformat_input), None) if options_key is None: app.logger.info(f"Options key not found for {hashformat_input}") abort(404) if request.method == 'GET': # Determine if this algorithm supports HMAC or salt options_key_lower = options_key.lower() # Log for debugging app.logger.info(f"Options key: {options_key}") # Determine support based on the algorithm # Algorithms using _hash_with_salt(): htpasswd-apache, des if options_key_lower in ['htpasswd-apache', 'des']: supports_hmac = False supports_salt = True # Algorithms using _hash_text_only(): adler32, blowfish, crc-32, crc-32b elif options_key_lower in ['adler32', 'blowfish', 'crc-32', 'crc-32b']: supports_hmac = False supports_salt = False # All others use _hash_with_hmac() else: supports_hmac = True supports_salt = False app.logger.info(f"supports_hmac: {supports_hmac}, supports_salt: {supports_salt}") # Get algorithm name for display algorithm_name = options_key.replace('-', ' ').title() return render_template('converter-hash.html', filetypes=available_hashtypes, filetype=filetype, fileformat=hashformat_input, options=json.dumps(hash_options[options_key], sort_keys=False, indent=2), tu=options_key, algorithm_name=algorithm_name, supports_hmac=supports_hmac, supports_salt=supports_salt) urls = request.form.getlist('url[]') if not urls: urls = [{"path": "empty"}] lang = getattr(g, 'lang_code', 'en') hash_lang_prefix = f"/{lang}" if lang != "en" else "" _batch_limit = get_user_batch_limit() if len(urls) > _batch_limit: flash(_("Sorry! we only support up to %(n)s files per batch.", n=_batch_limit), "error") return redirect(f"{hash_lang_prefix}/{hashformat_input}-generator") # ── Enforce daily hash limit ───────────────────────────────────────── _hash_user_id = session.get('user_id') _hash_ip = get_client_ip() _hash_limit = check_conversion_limit(user_id=_hash_user_id, ip_address=_hash_ip, tool_type='hash') if not _hash_limit['allowed']: flash( f"Daily hash limit reached ({_hash_limit['used']}/{_hash_limit['limit']}). " f"Upgrade your plan for more.", "error" ) return redirect(f"{hash_lang_prefix}/hash/{hashformat_input}-generator") # ──────────────────────────────────────────────────────────────────── _hash_args = (urls, options_key, json.loads(request.form.get('options') or '{}'), { "UPLOAD_DIR": app.config['UPLOAD_DIR'], "BUCKET": app.config['BUCKET'], "LOCAL": app.config['LOCAL'] }) _hash_queue_name = _get_job_queue_name() _hash_timeout = _get_plan_timeout_seconds() job = _local_queue.enqueue_call(func=convert_list[filetype].convert, args=_hash_args, result_ttl=5000, timeout=_hash_timeout, queue_name=_hash_queue_name) # ── Record hash usage ──────────────────────────────────────────────── try: _src_name = urls[0] if urls and urls[0] != '{"path": "empty"}' else 'text-input' if isinstance(_src_name, dict): _src_name = _src_name.get('name', 'hash-input') record_tool_usage('hash', user_id=_hash_user_id, ip_address=_hash_ip, file_name=str(_src_name)[:120], status='finished') except Exception: pass # ──────────────────────────────────────────────────────────────────── # Determine correct redirect based on subdomain current_subdomain = _filetype_from_host() if current_subdomain == 'hash' or current_subdomain == 'hash-generators': _host = request.host.split(':')[0] return redirect(f"{request.scheme}://{_host}{hash_lang_prefix}/hash-output/{job.id}") else: return redirect(f"{hash_lang_prefix}/hash-output/{job.id}") # ===== OUTPUT ROUTES (MAIN DOMAIN) ===== @app.route("//output/", methods=['GET']) @app.route('/output/', methods=['GET']) def output_route(job_id, lang_code=None): return output(job_id, lang_code=lang_code) @app.route("//output//", methods=['GET']) @app.route('/output//', methods=['GET']) def output_nondomain(filetype, job_id, lang_code=None): if filetype not in _OUTPUT_ALLOWED: abort(404) return output(job_id, lang_code=lang_code) @app.route("//hash-output/", subdomain='hash', methods=['GET']) @app.route('/hash-output/', subdomain='hash', methods=['GET']) def subdomain_hashoutput(job_id, lang_code=None): """Handle hash output on hash subdomain: hash.onlineconvert.cc/hash-output/job_id""" app.logger.info(f"===== SUBDOMAIN HASHOUTPUT HIT =====") app.logger.info(f"job_id: {job_id}, lang_code: {lang_code}") # No filetype parameter needed since we know it's the hash subdomain return hashoutput(job_id, lang_code=lang_code) @app.route("//hash-output/", methods=['GET']) @app.route('/hash-output/', methods=['GET']) def hashoutput_route(job_id, lang_code=None): app.logger.info(f"===== HASHOUTPUT_ROUTE HIT =====") app.logger.info(f"job_id: {job_id}, lang_code: {lang_code}") return hashoutput(job_id, lang_code=lang_code) @app.route('/retry/', methods=['POST']) def retry_job(job_id): """Re-queue a failed conversion using the original uploaded file.""" from helpers import is_admin as _is_admin db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as _rc: _rc.row_factory = sqlite3.Row conv = _rc.execute( "SELECT c.*, fu.file_path AS upload_file_path " "FROM conversions c " "LEFT JOIN file_uploads fu ON fu.id = c.upload_id " "WHERE c.job_id=?", (job_id,) ).fetchone() if not conv: flash("Conversion record not found.", "error") return redirect(url_for('output_route', job_id=job_id)) conv = dict(conv) if conv.get('status') != 'failed': flash("Only failed conversions can be retried.", "error") return redirect(url_for('output_route', job_id=job_id)) _uid = session.get('user_id') _admin = _is_admin() if not _admin: _conv_user = conv.get('user_id') if _conv_user: if str(_conv_user) != str(_uid): flash("You do not have permission to retry this conversion.", "error") return redirect(url_for('output_route', job_id=job_id)) else: _req_ip = get_client_ip() _conv_ip = conv.get('ip_address') or '' if _conv_ip and _req_ip != _conv_ip: flash("You do not have permission to retry this conversion.", "error") return redirect(url_for('output_route', job_id=job_id)) upload_file_path = conv.get('upload_file_path') if not upload_file_path: flash("Original file not available — please re-upload to convert again.", "error") return redirect(url_for('output_route', job_id=job_id)) full_disk_path = os.path.join(app.config['UPLOAD_DIR'], upload_file_path) if not os.path.exists(full_disk_path): flash("Original file is no longer available (it may have been cleaned up). Please re-upload to convert again.", "error") return redirect(url_for('output_route', job_id=job_id)) try: stored = json.loads(conv.get('options_json') or '{}') _filetype = stored.get('filetype') or conv.get('tool_type') or 'converter' _options = stored.get('options') or {} except Exception: _filetype = conv.get('tool_type') or 'converter' _options = {} if _filetype not in convert_list: flash("Cannot retry: original converter type is unknown or unavailable.", "error") return redirect(url_for('output_route', job_id=job_id)) _out_fmt = conv.get('output_format') or '' if not _out_fmt: flash("Could not determine output format for retry.", "error") return redirect(url_for('output_route', job_id=job_id)) _queue_name = conv.get('queue') or 'low' if _queue_name not in ('high', 'medium', 'low'): _queue_name = 'low' _retry_base, _retry_ext = os.path.splitext(os.path.basename(upload_file_path)) _url_entry = json.dumps({"path": upload_file_path, "upload_id": conv.get('upload_id'), "name": _retry_base, "ext": _retry_ext}) _conv_args = ([_url_entry], _out_fmt, _options, { "UPLOAD_DIR": app.config.get('UPLOAD_DIR', 'storage/uploads'), "BUCKET": app.config.get('BUCKET', ''), "LOCAL": app.config.get('LOCAL', True) }) try: _retry_timeout = _get_plan_timeout_seconds() new_job = _local_queue.enqueue_call( func=convert_list[_filetype].convert, args=_conv_args, result_ttl=5000, timeout=_retry_timeout, queue_name=_queue_name ) _options_json = json.dumps({"filetype": _filetype, "options": _options}) _conv_ip = get_client_ip() with sqlite3.connect(db_path) as _cc: _cc.execute( "INSERT INTO conversions (user_id, upload_id, input_format, output_format, job_id, status, file_name, ip_address, tool_type, queue, options_json) VALUES (?,?,?,?,?,?,?,?,?,?,?)", (conv.get('user_id'), conv.get('upload_id'), conv.get('input_format', ''), _out_fmt, new_job.id, 'pending', conv.get('file_name', ''), _conv_ip, conv.get('tool_type', 'converter'), _queue_name, _options_json) ) _cc.commit() return redirect(url_for('output_route', job_id=new_job.id)) except Exception as _re: flash(f"Retry failed: {_re}", "error") return redirect(url_for('output_route', job_id=job_id)) @app.route('/job/status/', methods=['GET']) @app.route("//job/status/", methods=['GET']) def jobStatus_route(job_id, lang_code=None): if lang_code and lang_code in supported_languages: g.lang_code = lang_code return jobStatus(job_id) # ===== JOB STATUS ROUTES (Subdomain) ===== @app.route('/job/status/', subdomain='', methods=['GET']) def subdomain_jobStatus(filetype, job_id): """Handle job status on subdomains: image.onlineconvert.cc/job/status/job_id""" app.logger.info(f"===== SUBDOMAIN JOB STATUS HIT =====") app.logger.info(f"filetype: {filetype}, job_id: {job_id}") return jobStatus(job_id) @app.route('//job/status/', subdomain='', methods=['GET']) def subdomain_jobStatus_lang(filetype, lang_code, job_id): """Handle job status on subdomains with language: image.onlineconvert.cc/zh/job/status/job_id""" app.logger.info(f"===== SUBDOMAIN JOB STATUS WITH LANG HIT =====") app.logger.info(f"filetype: {filetype}, lang: {lang_code}, job_id: {job_id}") if lang_code and lang_code in supported_languages: g.lang_code = lang_code return jobStatus(job_id) @app.route('/api/job//notify-email', subdomain='', methods=['POST']) def subdomain_api_job_notify_email(filetype, job_id): return api_job_notify_email(job_id) @app.route('//api/job//notify-email', subdomain='', methods=['POST']) def subdomain_api_job_notify_email_lang(filetype, lang_code, job_id): if lang_code and lang_code in supported_languages: g.lang_code = lang_code return api_job_notify_email(job_id) @app.route('/api/vapid-public-key', subdomain='', methods=['GET']) def subdomain_api_vapid_public_key(filetype): return api_vapid_public_key() @app.route('/api/job//notify-push', subdomain='', methods=['POST']) def subdomain_api_job_notify_push(filetype, job_id): return api_job_notify_push(job_id) @app.route('//api/job//notify-push', subdomain='', methods=['POST']) def subdomain_api_job_notify_push_lang(filetype, lang_code, job_id): if lang_code and lang_code in supported_languages: g.lang_code = lang_code return api_job_notify_push(job_id) @app.route('/api/push/subscribe-global', subdomain='', methods=['POST']) def subdomain_api_push_subscribe_global(filetype): return api_push_subscribe_global() @app.route('/api/push/global-status', subdomain='', methods=['GET']) def subdomain_api_push_global_status(filetype): return api_push_global_status() @app.route('/api/push/unsubscribe-global', subdomain='', methods=['POST']) def subdomain_api_push_unsubscribe_global(filetype): return api_push_unsubscribe_global() # ===== UPLOAD ROUTES ===== @app.route('/upload', methods=['POST'], subdomain="") @app.route('/upload', methods=['POST']) def upload(filetype=None): if filetype is None: filetype = _filetype_from_host() # Also accept filetype via query string (e.g. /upload?filetype=hash) if filetype is None: filetype = request.args.get('filetype') or request.form.get('filetype') allowed = [ "image", "audio", "video", "document", "device", "archive", "ebook", "webservice", "pdf", "bmp", "hash", "document-compressor", "image-compressor", "video-compressor" ] if filetype is not None and filetype not in allowed: abort(404) files = request.files.getlist('files') if not files: return jsonify(success=False, message="No files uploaded"), 400 user_id = session.get('user_id') sess_id = session.get('anon_session_id', '') if not sess_id and not user_id: sess_id = uuid.uuid4().hex session['anon_session_id'] = sess_id # ── Detect tool type from filetype ────────────────────────────────── _tool_type = 'hash' if filetype == 'hash' else ('pdf' if filetype == 'pdf' else 'converter') # ── Resolve team context (explicit team_id in form/query triggers team plan) ── _team_plan_id_override = None _team_id_ctx = None try: _team_id_ctx = int(request.form.get('team_id') or request.args.get('team_id') or 0) or None except (TypeError, ValueError): _team_id_ctx = None if _team_id_ctx and user_id: with _db() as _tc: _tm_row = _tc.execute( "SELECT 1 FROM team_members WHERE team_id=? AND user_id=? AND status='active' LIMIT 1", (_team_id_ctx, user_id) ).fetchone() if not _tm_row: _owner_row = _tc.execute( "SELECT 1 FROM teams WHERE id=? AND owner_user_id=?", (_team_id_ctx, user_id) ).fetchone() _tm_row = _owner_row if _tm_row: _t_row = _tc.execute( "SELECT plan_id FROM teams WHERE id=?", (_team_id_ctx,) ).fetchone() if _t_row and _t_row['plan_id']: _team_plan_id_override = _t_row['plan_id'] # ──────────────────────────────────────────────────────────────────── # ── Enforce daily conversion limit (team plan overrides personal plan when in team context) ── _client_ip = get_client_ip() _limit_info = check_conversion_limit( user_id=user_id, ip_address=_client_ip, tool_type=_tool_type, plan_id_override=_team_plan_id_override ) if not _limit_info['allowed']: return jsonify( success=False, limit_exceeded=True, message=( f"You have reached your daily limit of {_limit_info['limit']} conversion(s) " f"on the {_limit_info['plan_name']} plan. " f"Upgrade your plan to convert more files today." ), used=_limit_info['used'], limit=_limit_info['limit'], plan_name=_limit_info['plan_name'], upgrade_url='/pricing', ), 429 # ──────────────────────────────────────────────────────────────────── # ── Effective file-size limit (server cap vs plan cap; team plan when in team context) ── _srv_max_mb = app.config.get('MAX_CONTENT_LENGTH', 500 * 1024 * 1024) // (1024 * 1024) _upload_user = get_current_user() _upload_plan = ( get_plan_by_id(_team_plan_id_override) if _team_plan_id_override else (get_plan_by_id(_upload_user['plan_id']) if _upload_user else None) ) _plan_max_mb = int(_upload_plan['max_file_size_mb']) if ( _upload_plan and _upload_plan.get('max_file_size_mb') and int(_upload_plan['max_file_size_mb']) > 0 ) else None # Effective limit: use plan limit if set, but never exceed server ceiling if _plan_max_mb is not None: _effective_max_mb = min(_srv_max_mb, _plan_max_mb) else: _effective_max_mb = _srv_max_mb _effective_max_bytes = _effective_max_mb * 1024 * 1024 # ──────────────────────────────────────────────────────────────────── uploaded_files = [] os.makedirs(app.config['UPLOAD_DIR'], exist_ok=True) for file in files: filename = secure_filename(file.filename) name, ext = os.path.splitext(filename) ext = ext.lower() folder = uuid.uuid4().hex folder_path = os.path.join(app.config['UPLOAD_DIR'], folder) os.makedirs(folder_path, exist_ok=True) save_path = os.path.join(folder_path, filename) file.save(save_path) file_size = os.path.getsize(save_path) # Server-side size enforcement (plan limit vs server ceiling) if file_size > _effective_max_bytes: try: os.remove(save_path) os.rmdir(folder_path) except OSError: pass return jsonify( success=False, message=f"File '{filename}' exceeds the {_effective_max_mb} MB limit for your plan." ), 413 file_path_rel = f"{folder}/{filename}" try: with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as _uc: cursor = _uc.execute( "INSERT INTO file_uploads (user_id, session_id, file_path, file_name, file_size, original_format) VALUES (?,?,?,?,?,?)", (user_id, sess_id, file_path_rel, filename, file_size, ext.lstrip('.')) ) upload_id = cursor.lastrowid _uc.commit() except Exception: upload_id = None uploaded_files.append({ "path": file_path_rel, "name": name, "ext": ext, "upload_id": upload_id }) return jsonify(success=True, files=uploaded_files) @app.route('/upload/remove', methods=['GET'], subdomain="") @app.route('/upload/remove', methods=['GET']) def remove_file(filetype=None): if filetype is None: filetype = _filetype_from_host() allowed = [ "image", "audio", "video", "document", "device", "archive", "ebook", "webservice", "pdf", "bmp", "hash", "document-compressor", "image-compressor", "video-compressor" ] if filetype is not None and filetype not in allowed: abort(404) path = request.args.get('fileid') if path: file_path = os.path.join(app.config['UPLOAD_DIR'], path) deleted_physical = False if os.path.exists(file_path): os.remove(file_path) deleted_physical = True # remove the now-empty parent folder parent_dir = os.path.dirname(file_path) try: if os.path.isdir(parent_dir) and not os.listdir(parent_dir): shutil.rmtree(parent_dir) except Exception: pass # always clean up the DB record, even if the file was already gone try: db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(db_path) as _db_conn: row = _db_conn.execute( "SELECT id FROM file_uploads WHERE file_path = ?", (path,) ).fetchone() if row: upload_id = row[0] _db_conn.execute( "DELETE FROM conversions WHERE upload_id = ?", (upload_id,) ) _db_conn.execute( "DELETE FROM file_uploads WHERE id = ?", (upload_id,) ) except Exception: pass if deleted_physical: return "ok" return "file not found", 404 @app.route('/pdf/save-result', methods=['POST']) @app.route('//pdf/save-result', methods=['POST']) def pdf_save_result(lang_code=None): file = request.files.get('file') if not file: return jsonify(success=False, message="No file"), 400 tool = request.form.get('tool', 'pdf-tool') original_name = request.form.get('original_name', '') filename = secure_filename(file.filename or 'result.pdf') folder = uuid.uuid4().hex folder_path = os.path.join(app.config['UPLOAD_DIR'], folder) os.makedirs(folder_path, exist_ok=True) save_path = os.path.join(folder_path, filename) file.save(save_path) file_size = os.path.getsize(save_path) file_path_rel = f"{folder}/{filename}" user_id = session.get('user_id') sess_id = session.get('anon_session_id', '') if not sess_id and not user_id: sess_id = uuid.uuid4().hex session['anon_session_id'] = sess_id try: ext_out = filename.rsplit('.', 1)[-1].lower() if '.' in filename else 'pdf' ext_in = original_name.rsplit('.', 1)[-1].lower() if '.' in original_name else 'pdf' _pdf_ip = get_client_ip() with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as _sc: _sc.execute( "INSERT INTO file_uploads (user_id, session_id, file_path, file_name, file_size, original_format) VALUES (?,?,?,?,?,?)", (user_id, sess_id, file_path_rel, filename, file_size, ext_out) ) upload_id = _sc.execute("SELECT last_insert_rowid()").fetchone()[0] _sc.execute( "INSERT INTO conversions (user_id, upload_id, input_format, output_format, job_id, status, file_name, completed_at, output_path, ip_address, tool_type) VALUES (?,?,?,?,?,?,?,datetime('now'),?,?,?)", (user_id, upload_id, ext_in, ext_out, f"client-{uuid.uuid4().hex[:8]}", 'finished', original_name, file_path_rel, _pdf_ip, 'pdf') ) _sc.commit() except Exception: pass return jsonify(success=True, path=file_path_rel) # ===== URL PREFETCH (Feature B: unified upload progress UI) ===== def _get_allowed_exts(ft, src_fmt=''): """Shared helper: return the set of lowercase allowed source extensions for a converter context. This is the single source of truth used by api_url_prefetch so the logic is not duplicated inline. ft – filetype key (e.g. 'image', 'pdf'). May be empty. src_fmt – explicit source format on pair pages (e.g. 'jpg'). May be empty. Returns a list; empty list means no restriction (unknown context).""" if src_fmt: return [src_fmt.lower()] if ft and ft in available_filetypes: return [f.lower() for f in available_filetypes[ft].get('allowed', [])] return [] def _filetype_from_referer(referer): """Derive (filetype, source_format) from the Referer header as a best-effort fallback. NOTE: Referer is NOT a security control — browsers may omit it (HTTPS downgrade, Referrer-Policy header, private browsing, etc.) and it must not be relied upon for authoritative validation. Use the HMAC-signed context token (url_prefetch_token) as the trusted source instead. Returns (filetype, source_format) — either may be an empty string when the referer does not match a known converter URL pattern.""" if not referer: return '', '' try: from urllib.parse import urlparse as _urlparse2 path = _urlparse2(referer).path.strip('/') parts = path.split('/') # Strip optional two-letter language prefix (e.g. /en/image/jpg) if parts and len(parts[0]) == 2 and parts[0].isalpha(): parts = parts[1:] if not parts: return '', '' segment = parts[0].lower() # Pattern: /{filetype}/... — category page (e.g. /image/jpg) if segment in available_filetypes: return segment, '' # Pattern: /{source}-to-{target} — pair page (e.g. /jpg-to-png) if '-to-' in segment: src, _, tgt = segment.partition('-to-') ft = find_filetype_for_formats(src, tgt) if ft: return ft, src except Exception: pass return '', '' @app.route('/api/url-prefetch', methods=['POST']) @app.route('/api/url-prefetch', subdomain='', methods=['POST']) def api_url_prefetch(filetype=None): """Download a remote URL to UPLOAD_DIR and return the local path. The frontend uses this so URL/cloud uploads show the same animated progress-bar
  • as regular PC file uploads.""" data = request.get_json(silent=True) or {} raw_url = (data.get('url') or '').strip() if not raw_url or not raw_url.lower().startswith(('http://', 'https://')): return jsonify({'ok': False, 'error': 'Invalid URL'}), 400 # ── Derive converter context — three-tier priority, fail-closed ───── # Context resolution order (most to least trusted): # 1. Subdomain route param: set by Flask URL routing, unfakeable. # 2. HMAC-signed ctx token: rendered server-side with app.secret_key; # tamper-proof — forging requires the secret key. # 3. Referer header: best-effort only (NOT a security control; may be # omitted by Referrer-Policy, HTTPS downgrade, privacy mode, etc.). # Client-supplied filetype/source_format JSON fields are NEVER used. # # IMPORTANT — fail-closed: if context cannot be resolved by any of the # above, the upload is REJECTED (HTTP 422) rather than allowed through # without validation. This prevents a bypass where a caller omits both # ctx and Referer to circumvent the extension check. # # NOTE: ctx is ALWAYS parsed when present, even if the subdomain param # already provides ft. Pair pages (e.g. /jpg-to-png on the image # subdomain) embed src_fmt in the token; skipping ctx parse would fall # back to category-wide allowed extensions instead of [src_fmt] only. import hmac as _hmac_mod, hashlib as _hashlib_mod ft, src_fmt = filetype or '', '' ctx_resolved = bool(ft) # truthy when subdomain route param was provided ctx = (data.get('ctx') or '').strip() if ctx: try: parts = ctx.rsplit(':', 1) if len(parts) == 2: ctx_body, ctx_sig = parts expected = _hmac_mod.new( app.secret_key.encode() if isinstance(app.secret_key, str) else app.secret_key, ctx_body.encode(), _hashlib_mod.sha256 ).hexdigest() if _hmac_mod.compare_digest(ctx_sig, expected): ctx_ft, ctx_sf = ctx_body.split(':', 1) if ':' in ctx_body else (ctx_body, '') if ft: # Subdomain already set ft — verify token ft matches; # reject on mismatch (token from a different converter). if ctx_ft and ctx_ft != ft: return jsonify({ 'ok': False, 'error': 'Converter context mismatch.' }), 422 else: ft = ctx_ft ctx_resolved = True # Always take src_fmt from token so pair-page restriction # (src_fmt=[one format]) works on subdomain routes too. src_fmt = ctx_sf except Exception: pass if not ctx_resolved: ft_r, sf_r = _filetype_from_referer(request.referrer or '') if ft_r: ft, src_fmt = ft_r, sf_r ctx_resolved = True if not ctx_resolved: return jsonify({ 'ok': False, 'error': 'Converter context could not be determined. ' 'Please upload from a converter page.' }), 422 # ───────────────────────────────────────────────────────────────────── try: _dl_errors = [] resolved = _resolve_urls([raw_url], app.config['UPLOAD_DIR'], _errors=_dl_errors) if not resolved: reason = _dl_errors[0] if _dl_errors else 'Download failed' return jsonify({'ok': False, 'error': reason}), 500 entry_str = resolved[0] try: entry = json.loads(entry_str) rel_path = entry.get('path', '') except Exception: rel_path = entry_str full_path = os.path.join(app.config['UPLOAD_DIR'], rel_path) filename = os.path.basename(rel_path) # ── File-type validation ────────────────────────────────────────── ext = os.path.splitext(filename)[1].lstrip('.').lower() if ext: allowed_exts = _get_allowed_exts(ft, src_fmt) if allowed_exts and ext not in allowed_exts: # Remove the just-downloaded file before returning the error try: import shutil as _shutil _shutil.rmtree(os.path.dirname(full_path), ignore_errors=True) except Exception: pass return jsonify({ 'ok': False, 'error': f'File type .{ext} is not allowed on this converter' }), 422 # ───────────────────────────────────────────────────────────────── size = os.path.getsize(full_path) units = ['B', 'KB', 'MB', 'GB'] sz, ui = float(size), 0 while sz >= 1024 and ui < len(units) - 1: sz /= 1024; ui += 1 size_human = f'{sz:.1f} {units[ui]}' return jsonify({'ok': True, 'path': rel_path, 'filename': filename, 'size': size, 'size_human': size_human}) except Exception as e: app.logger.error(f'url-prefetch error: {e}') return jsonify({'ok': False, 'error': str(e)}), 500 # ===== EXPORT ARCHIVE (Feature A: batch archive download) ===== @app.route('/api/job//export-archive', methods=['POST']) @app.route('/job//export-archive', methods=['POST']) def api_job_export_archive(job_id): """Pack all output files for a finished job into a chosen archive format and stream it back. Supported formats: zip, tar.gz, tar.bz2, 7z. Requires the user's plan to have batch_archive_download=1. The job must belong to the requesting user/session (IDOR guard).""" from converters.archive_converter import convert as _archive_convert from flask import Response as _Response, send_file as _send_file # ── Plan gate ────────────────────────────────────────────────────────── _user = get_current_user() _plan = get_plan_by_id(_user['plan_id']) if _user else get_plan_by_id(1) if not _plan or not int(_plan.get('batch_archive_download', 0)): return jsonify({'ok': False, 'error': 'Your plan does not include batch archive downloads.'}), 403 # ── Archive format ───────────────────────────────────────────────────── data = request.get_json(silent=True) or {} fmt = (data.get('format') or 'zip').lower().strip('.') if fmt not in ('zip', 'tar.gz', 'tar.bz2', '7z'): fmt = 'zip' # ── Ownership check (IDOR guard) ─────────────────────────────────────── _user_id = session.get('user_id') try: _db_path = os.path.join(dir_path, 'storage', 'sqlite.db') with sqlite3.connect(_db_path) as _aconn: _aconn.row_factory = sqlite3.Row _crow = _aconn.execute( "SELECT user_id, ip_address FROM conversions WHERE job_id=? LIMIT 1", (job_id,) ).fetchone() if _crow is None: return jsonify({'ok': False, 'error': 'Job not found.'}), 404 _job_user = _crow['user_id'] # Authenticated users: must own the job if _user_id and _job_user and int(_job_user) != int(_user_id): return jsonify({'ok': False, 'error': 'Access denied.'}), 403 # Anonymous jobs (no user_id in DB): allow — job_id is a UUID except Exception: # Fail-closed: if ownership cannot be verified, deny access return jsonify({'ok': False, 'error': 'Could not verify job ownership.'}), 403 # ── Job result ───────────────────────────────────────────────────────── try: job = LocalJob.fetch(job_id) except Exception: return jsonify({'ok': False, 'error': 'Job not found.'}), 404 if not job.is_finished or not job.result: return jsonify({'ok': False, 'error': 'Job is not finished.'}), 400 result = job.result if isinstance(job.result, dict) else {} file_paths = result.get('results') or [] if not file_paths: return jsonify({'ok': False, 'error': 'No output files found.'}), 400 # ── Build URL entries for archive_converter ──────────────────────────── upload_dir = app.config['UPLOAD_DIR'] url_entries = [] for fp in file_paths: if not fp: continue if os.path.isabs(fp): full = fp rel = os.path.relpath(fp, upload_dir) else: rel = fp full = os.path.join(upload_dir, fp) if not os.path.isfile(full): continue basename = os.path.basename(full) name, ext = os.path.splitext(basename) url_entries.append(json.dumps({'path': rel, 'name': name, 'ext': ext})) if not url_entries: return jsonify({'ok': False, 'error': 'No accessible output files found.'}), 400 # ── Run archive_converter ────────────────────────────────────────────── arc_result = _archive_convert( url_entries, fmt, {}, {'UPLOAD_DIR': upload_dir} ) if arc_result.get('error'): return jsonify({'ok': False, 'error': arc_result.get('message', 'Archive creation failed.')}), 500 arc_path = arc_result.get('output_path') or (arc_result.get('results') or [None])[0] if not arc_path or not os.path.isfile(arc_path): return jsonify({'ok': False, 'error': 'Archive file not found after creation.'}), 500 # ── MIME type mapping ────────────────────────────────────────────────── mime_map = { 'zip': 'application/zip', 'tar.gz': 'application/gzip', 'tar.bz2': 'application/x-bzip2', '7z': 'application/x-7z-compressed', } mimetype = mime_map.get(fmt, 'application/octet-stream') dl_name = f'converted_{job_id[:8]}.{fmt}' return _send_file(arc_path, mimetype=mimetype, as_attachment=True, download_name=dl_name) # ===== RECORD USAGE (client-side tools: PDF, etc.) ===== @app.route('/record-usage', methods=['POST']) def record_usage(): """Lightweight endpoint for client-side tools to record usage without uploading a file.""" tool_type = request.json.get('tool_type', 'pdf') if request.is_json else request.form.get('tool_type', 'pdf') file_name = (request.json.get('file_name', '') if request.is_json else request.form.get('file_name', ''))[:120] _user_id = session.get('user_id') _ip = get_client_ip() try: record_tool_usage(tool_type, user_id=_user_id, ip_address=_ip, file_name=file_name, status='finished') except Exception: pass return jsonify(success=True) # ===== EXPORT ARCHIVE (Subdomain mirrors) ===== @app.route('/api/job//export-archive', subdomain='', methods=['POST']) @app.route('/job//export-archive', subdomain='', methods=['POST']) def subdomain_api_job_export_archive(filetype, job_id): """Archive download on subdomains — delegates to main handler so the session (and plan) is resolved against the subdomain the user is on.""" return api_job_export_archive(job_id) # ===== SUBDOMAIN CATCH-ALL (must be before main domain catch-all) ===== @app.route('/', methods=['GET', 'POST'], subdomain="") def convert_to(filetype, subpath): """Catch-all for subdomain paths not matched by specific routes""" app.logger.info(f"===== CONVERT_TO ROUTE HIT =====") app.logger.info(f"filetype: {filetype}") app.logger.info(f"subpath: {subpath}") return _convert_to_handler(filetype.lower(), subpath) # ===== BARE /first-to-second SHORT-FORM REDIRECT ===== # e.g. onlineconvert.cc/jpg-to-bmp → image.onlineconvert.cc/convert/jpg-to-bmp # This is a fallback for when the before_request subdomain hooks don't fire. @app.route('/-to-', methods=['GET']) def short_format_to_format(first, second): """onlineconvert.cc/jpg-to-bmp — redirect to the correct filetype subdomain.""" first = first.lower() second = second.lower() lang_prefix = f"/{g.lang_code}" if getattr(g, 'lang_code', 'en') != 'en' else "" filetype = find_filetype_for_formats(first, second) if filetype and _is_subdomains_enabled(): server_name = app.config.get('SERVER_NAME') or website_url return redirect( f"{request.scheme}://{filetype}.{server_name}{lang_prefix}/convert/{first}-to-{second}", code=301 ) # Subdomains disabled or unknown pair — serve the converter page directly if filetype: return render_converter_page(filetype, second, first) abort(404) # ===== MAIN DOMAIN CATCH-ALL (must be last) ===== @app.route('/', methods=['GET', 'POST']) def catch_all(subpath): """Handle any unmatched routes on main domain""" abort(404)