main.py

# encoding = utf-8
import threading
import time
import logging
import json
import socket
import re
import sys
import os
from pathlib import Path
from urllib import error as urllib_error
from tkinter.messagebox import showerror
from tkinter import *
from tkinter.ttk import *


_STDIO_FALLBACK_BINARIES = []
_STDERR_DISCARDED = False


class _DevNullConsoleStream:
    """A minimal text stream proxy that exposes a stable .buffer."""

    def __init__(self, binary):
        self.buffer = binary
        self.encoding = "utf-8"

    def write(self, data):
        if isinstance(data, str):
            data = data.encode("utf-8", errors="ignore")
        elif not isinstance(data, (bytes, bytearray)):
            data = str(data).encode("utf-8", errors="ignore")
        try:
            self.buffer.write(data)
        except Exception:
            return 0
        return len(data)

    def flush(self):
        try:
            self.buffer.flush()
        except Exception:
            pass

    def isatty(self):
        return False


def _is_working_stream(stream) -> bool:
    if stream is None or getattr(stream, "closed", False):
        return False
    try:
        stream.write("")
        stream.flush()
    except Exception:
        return False
    return True


def _usable_buffered_stream(*candidates):
    for stream in candidates:
        if stream is None:
            continue
        if hasattr(stream, "buffer") and _is_working_stream(stream):
            return stream
    return None


def _make_devnull_text_stream():
    binary = open(os.devnull, "wb", buffering=0)
    _STDIO_FALLBACK_BINARIES.append(binary)
    return _DevNullConsoleStream(binary)


def ensure_stdio_for_headless_parent_console():
    """Discard stdio output when parent console stream is unavailable."""
    global _STDERR_DISCARDED

    stdout = _usable_buffered_stream(sys.stdout, getattr(sys, "__stdout__", None))
    if stdout is None:
        sys.stdout = _make_devnull_text_stream()
    else:
        sys.stdout = stdout

    stderr = _usable_buffered_stream(sys.stderr, getattr(sys, "__stderr__", None))
    if stderr is None:
        sys.stderr = _make_devnull_text_stream()
        _STDERR_DISCARDED = True
    else:
        sys.stderr = stderr
        _STDERR_DISCARDED = False


ensure_stdio_for_headless_parent_console()

from you_get import common

# import widgets
import main_window
import multiplatform
from i18n import I18n
from eventqueue import EventQueueForTkinter
debug = True


class MergedProgressEventQueueForTkinter(EventQueueForTkinter):
    """Coalesce frequent progress updates to keep only the latest snapshot."""

    def __init__(self, tk, progress_callback, delay_ms=100):
        super().__init__(tk=tk, delay_ms=delay_ms)
        self._progress_callback = progress_callback
        self._progress_lock = threading.Lock()
        self._progress_latest = None
        self._progress_flush_queued = False

    def add_progress_event(self, value: float, piece: str = "", speed: str = "", received_fraction: str = ""):
        payload = {
            "value": float(value),
            "piece": piece,
            "speed": speed,
            "received_fraction": received_fraction,
        }
        should_schedule_flush = False
        with self._progress_lock:
            self._progress_latest = payload
            if not self._progress_flush_queued:
                self._progress_flush_queued = True
                should_schedule_flush = True

        if should_schedule_flush:
            super().add_event(self._flush_progress_event)

    def _flush_progress_event(self):
        payload = None
        with self._progress_lock:
            payload = self._progress_latest
            self._progress_latest = None
            self._progress_flush_queued = False

        if payload is None:
            return

        self._progress_callback(
            payload["value"],
            piece=payload["piece"],
            speed=payload["speed"],
            received_fraction=payload["received_fraction"],
        )

        with self._progress_lock:
            has_new_payload = self._progress_latest is not None
            if has_new_payload and not self._progress_flush_queued:
                self._progress_flush_queued = True
        if has_new_payload:
            super().add_event(self._flush_progress_event)


def setup_logging() -> Path:
    """Configure root logging to console and a reset-on-start .log file."""
    log_path = Path(".log")
    level = logging.DEBUG if debug else logging.INFO
    log_format = '[%(asctime)s] [%(levelname)s] %(message)s'

    root_logger = logging.getLogger()
    root_logger.setLevel(level)
    # Clear inherited/default handlers to avoid duplicate logs when re-run.
    root_logger.handlers.clear()

    file_handler = logging.FileHandler(log_path, mode="w", encoding="utf-8")
    file_handler.setLevel(level)
    file_handler.setFormatter(logging.Formatter(log_format))

    root_logger.addHandler(file_handler)
    if not _STDERR_DISCARDED:
        console_handler = logging.StreamHandler()
        console_handler.setLevel(level)
        console_handler.setFormatter(logging.Formatter(log_format))
        root_logger.addHandler(console_handler)

    return log_path


LOG_FILE_PATH = setup_logging()

DEFAULT_CONFIG = {
    "download_dir": str(multiplatform.get_default_download_dir()),
    "language": "zh_CN",
    "download_options": {
        "download_captions": True,
        "merge_video_parts": True,
        "download_m3u8_video": False,
        "auto_rename": True,
        "ignore_ssl_errors": False,
        "forced_download": False,
        "skip_downloaded_video": False,
        "video_password_power": False,
        "video_password_value": "",
        "open_dir_after_finish": True,
        "debug": False,
    },
    "proxy_options": {
        "proxy_power": False,
        "proxy_type": "Socks5",
        "proxy_host": "",
        "proxy_login_power": False,
        "proxy_login_value": "",
    },
}


_PROXY_ENDPOINT_RE = re.compile(
    r"^\s*(?:(?P<scheme>[a-zA-Z][a-zA-Z0-9+.-]*)://)?(?P<authority>[^\s/?#]+)"
)


def normalize_proxy_endpoint(proxy_input: str, fallback_scheme: str) -> tuple[str, str]:
    """Normalize user proxy input into (scheme, authority_without_scheme)."""
    raw = str(proxy_input or "").strip()
    if not raw:
        return fallback_scheme, ""

    match = _PROXY_ENDPOINT_RE.match(raw)
    if not match:
        return fallback_scheme, raw

    detected_scheme = (match.group("scheme") or "").lower()
    authority = (match.group("authority") or "").lstrip("/")

    if detected_scheme.startswith("socks"):
        scheme = "socks5"
    elif detected_scheme:
        scheme = "http"
    else:
        scheme = fallback_scheme

    return scheme, authority


def normalize_config(raw_config: dict) -> dict:
    config = {
        "download_dir": DEFAULT_CONFIG["download_dir"],
        "language": DEFAULT_CONFIG["language"],
        "download_options": dict(DEFAULT_CONFIG["download_options"]),
        "proxy_options": dict(DEFAULT_CONFIG["proxy_options"]),
    }
    if not isinstance(raw_config, dict):
        return config

    if "download_dir" in raw_config:
        config["download_dir"] = str(raw_config.get("download_dir") or DEFAULT_CONFIG["download_dir"])
    if "language" in raw_config:
        config["language"] = str(raw_config.get("language") or DEFAULT_CONFIG["language"])

    raw_download_options = raw_config.get("download_options", {})
    if isinstance(raw_download_options, dict):
        for key in config["download_options"].keys():
            if key in raw_download_options:
                config["download_options"][key] = raw_download_options[key]

    raw_proxy_options = raw_config.get("proxy_options", {})
    if isinstance(raw_proxy_options, dict):
        for key in config["proxy_options"].keys():
            if key in raw_proxy_options:
                config["proxy_options"][key] = raw_proxy_options[key]

    return config


def build_download_kwargs(config: dict) -> dict:
    config = normalize_config(config)
    dl_opts = config["download_options"]
    proxy_opts = config["proxy_options"]

    output_dir = Path(config["download_dir"]).expanduser()
    kwargs = {
        "output_dir": output_dir,
        "merge": bool(dl_opts.get("merge_video_parts", True)),
        "info_only": False,
        "json_output": False,
        "caption": bool(dl_opts.get("download_captions", True)),
    }

    if bool(dl_opts.get("video_password_power", False)):
        password = str(dl_opts.get("video_password_value", "")).strip()
        if password:
            kwargs["password"] = password

    proxy_power = bool(proxy_opts.get("proxy_power", False))
    proxy_host = str(proxy_opts.get("proxy_host", "") or "").strip()
    if proxy_power and proxy_host:
        proxy_type = str(proxy_opts.get("proxy_type", "Socks5"))
        login_power = bool(proxy_opts.get("proxy_login_power", False))
        login_value = str(proxy_opts.get("proxy_login_value", "") or "").strip()
        fallback_scheme = "socks5" if proxy_type.lower().startswith("socks") else "http"
        scheme, proxy_auth_host = normalize_proxy_endpoint(proxy_host, fallback_scheme)
        if login_power and login_value:
            if "@" not in proxy_auth_host:
                proxy_auth_host = f"{login_value}@{proxy_auth_host}"
        kwargs["extractor_proxy"] = f"{scheme}://{proxy_auth_host}"

    return kwargs


def apply_common_settings(config: dict):
    config = normalize_config(config)
    dl_opts = config["download_options"]

    if "download_captions" in dl_opts:
        common.caption = bool(dl_opts.get("download_captions"))
    if "auto_rename" in dl_opts:
        common.auto_rename = bool(dl_opts.get("auto_rename", common.auto_rename))
    if "ignore_ssl_errors" in dl_opts:
        common.insecure = bool(dl_opts.get("ignore_ssl_errors", common.insecure))
    if "download_m3u8_video" in dl_opts:
        common.m3u8 = bool(dl_opts.get("download_m3u8_video", common.m3u8))
    if "forced_download" in dl_opts:
        common.force = bool(dl_opts.get("forced_download", common.force))
    if "skip_downloaded_video" in dl_opts:
        common.skip_existing_file_size_check = bool(
            dl_opts.get("skip_downloaded_video", common.skip_existing_file_size_check)
        )

    proxy_opts = config["proxy_options"]
    proxy_power = bool(proxy_opts.get("proxy_power", False))
    proxy_host = str(proxy_opts.get("proxy_host", "") or "").strip()

    if not proxy_power or not proxy_host:
        common.extractor_proxy = None
        if hasattr(common, "set_http_proxy"):
            common.set_http_proxy("")
        return

    proxy_type = str(proxy_opts.get("proxy_type", "Socks5"))
    login_power = bool(proxy_opts.get("proxy_login_power", False))
    login_value = str(proxy_opts.get("proxy_login_value", "") or "").strip()
    use_socks = proxy_type.lower().startswith("socks")

    fallback_scheme = "socks5" if use_socks else "http"
    scheme, proxy_auth_host = normalize_proxy_endpoint(proxy_host, fallback_scheme)
    if login_power and login_value and "@" not in proxy_auth_host:
        proxy_auth_host = f"{login_value}@{proxy_auth_host}"

    use_socks = scheme.startswith("socks")
    common.extractor_proxy = f"{scheme}://{proxy_auth_host}"

    try:
        if use_socks and hasattr(common, "set_socks_proxy"):
            common.set_socks_proxy(proxy_auth_host)
        elif hasattr(common, "set_http_proxy"):
            common.set_http_proxy(proxy_auth_host)
    except Exception:
        logging.exception("Failed to apply proxy settings to you_get.common")

    if dl_opts.get("debug"):
        logging.getLogger().setLevel(logging.DEBUG)


def load_config(apply_common: bool = True):
    config_path = Path("config.json")
    if config_path.exists():
        try:
            with open(config_path, "r", encoding="utf-8") as f:
                config = normalize_config(json.load(f))
        except Exception as e:
            logging.error("Failed to load config.json", exc_info=True)
            config = normalize_config({})
    else:
        logging.info("No config.json found, using defaults")
        config = normalize_config({})
    if apply_common:
        try:
            apply_common_settings(config)
        except Exception:
            logging.exception("Failed while applying config to you_get.common")

    return config
if __name__ == '__main__':
    logging.info("Application logging initialized: %s", LOG_FILE_PATH.resolve())
    # load configuration and apply to you_get.common
    cfg = load_config()
    i18n = I18n()
    language = str(cfg.get("language") or DEFAULT_CONFIG["language"])
    try:
        i18n.set_language(language)
    except ValueError:
        logging.warning("Unsupported language in config: %s, fallback to %s", language, i18n.default_lang)
        i18n.set_language(i18n.default_lang)

    app = main_window.MainWindow()
    is_downloading = threading.Event()
    app_queue = MergedProgressEventQueueForTkinter(tk=app, progress_callback=app.set_process, delay_ms=100)
    app_queue()

    def set_status_safe(text: str):
        app.status_label.config(text=text)

    def finish_download(status_text: str):
        if is_downloading.is_set():
            is_downloading.clear()
        app.url_b.config(state=NORMAL)
        app.status_label.config(text=status_text)

    def open_download_dir_if_needed(config: dict):
        cfg = normalize_config(config)
        if not bool(cfg["download_options"].get("open_dir_after_finish", False)):
            return
        try:
            folder = Path(cfg["download_dir"]).expanduser()
            folder.mkdir(parents=True, exist_ok=True)
            multiplatform.open_folder(str(folder))
        except Exception:
            logging.exception("Failed to open download folder")

    def format_error_message(exc: Exception) -> str:
        failed_prefix = I18n().get("main_window.download_status.failed")
        if isinstance(exc, socket.timeout):
            return f"{failed_prefix}：网络超时"
        if isinstance(exc, urllib_error.HTTPError):
            return f"{failed_prefix}：HTTP {exc.code}"
        if isinstance(exc, urllib_error.URLError):
            reason = getattr(exc, "reason", "网络错误")
            return f"{failed_prefix}：{reason}"
        if isinstance(exc, (PermissionError, FileNotFoundError, OSError)):
            return f"{failed_prefix}：文件系统错误"
        if isinstance(exc, AssertionError):
            return f"{failed_prefix}：资源校验失败"
        return f"{failed_prefix}：{exc.__class__.__name__}"

    def format_system_exit_message(exc: SystemExit) -> str:
        failed_prefix = I18n().get("main_window.download_status.failed")
        code = getattr(exc, "code", None)
        if code in (None, 0):
            return f"{failed_prefix}：任务已终止"
        return f"{failed_prefix}：任务异常终止（exit={code}）"

    def download_worker(url: str):
        try:
            runtime_cfg = load_config(apply_common=True)
            common.any_download(url, **build_download_kwargs(runtime_cfg))
        except SystemExit as exc:
            # you-get 内部会通过 log.wtf() 调用 sys.exit()；在线程中需要显式拦截
            # 以避免线程异常退出且界面状态无法恢复。
            logging.exception("Download aborted by SystemExit for url=%s", url)
            failed_text = format_system_exit_message(exc)
            app_queue.add_event(lambda: finish_download(I18n().get("main_window.download_status.failed")))
            app_queue.add_event(showerror, I18n().get("errors.error_prompt").format(error_message=failed_text))
        except Exception as exc:
            logging.exception("Download failed for url=%s", url)
            failed_text = format_error_message(exc)
            app_queue.add_event(lambda: finish_download(I18n().get("main_window.download_status.failed")))
            app_queue.add_event(showerror, I18n().get("errors.error_prompt").format(error_message=failed_text))
        else:
            app_queue.add_event(lambda: finish_download(I18n().get("main_window.download_status.completed")))
            open_download_dir_if_needed(runtime_cfg)

    @app.bind_download
    def download_callback():
        if not is_downloading.is_set():
            is_downloading.set()
            app.url_b.config(state=DISABLED)
            set_status_safe(I18n().get("main_window.download_status.extracting"))
            threading.Thread(target=download_worker, args=(app.url_e.get(),), daemon=True).start()

    class NewProcessBar(common.SimpleProgressBar):
        def __init__(self, total_size, total_pieces=1):
            try:
                super().__init__(total_size, total_pieces)
            except TypeError:
                pass
            self.total_size = float(total_size) if total_size else 0.0
            self.total_pieces = int(total_pieces) if total_pieces else 1
            self._received = 0.0
            self._pieces_done = 0
            self._last_updated = time.time()
            self._speed = "0 B/s"
            # self.done_callback = None

        def update(self):
            # compute progress in 0.0-1.0, prefer byte-based when available
            progress = 0.0
            if self.total_size and self.total_size > 0:
                progress = min(1.0, float(self._received) / float(self.total_size))
                received_fraction = f"({self.format_byte(self._received)}/{self.format_byte(self.total_size)})" if self.total_size else ""
                app_queue.add_progress_event(
                    progress,
                    piece=f"[{self._pieces_done}/{self.total_pieces}]",
                    speed=self._speed,
                    received_fraction=received_fraction,
                )
            else:
                # fall back to piece-based progress
                progress = min(1.0, float(self._pieces_done) / max(1, self.total_pieces))
                app_queue.add_progress_event(progress, piece=f"[{self._pieces_done}/{self.total_pieces}]")

        def format_byte(self, bytes):
            # Simple function to format bytes into a human-readable string
            if bytes < 1024:
                return f"{bytes:.2f} B"
            elif bytes < 1024**2:
                return f"{bytes / 1024:.2f} KB"
            elif bytes < 1024**3:
                return f"{bytes / 1024**2:.2f} MB"
            else:
                return f"{bytes / 1024**3:.2f} GB"

        def update_received(self, n):
            self._received += float(n)
            time_diff= time.time() - self._last_updated
            byte_ps = n / time_diff if time_diff else 0.0
            self._speed = self.format_byte(byte_ps)+"/s"
            self._last_updated = time.time()
            self.update()

        def update_piece(self, n:int):
            self._pieces_done = n

        def done(self):
            pass
                
    common.SimpleProgressBar = NewProcessBar
    common.PiecesProgressBar = NewProcessBar

    app.pack()
    app.mainloop()