front.py

import json, time, re, urllib.parse, urllib.request, urllib.error, html as html_lib, threading, uuid
from concurrent.futures import ThreadPoolExecutor, wait, as_completed, TimeoutError as FuturesTimeoutError
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
import hashlib
import uuid

PORT = 7860
OPENER = urllib.request.build_opener()
FAST_WORKERS = 3
FULL_WORKERS = 48
FETCH_TIMEOUT = 1.6
FAST_SEARCH_TIMEOUT = 5.0
FULL_SEARCH_TIMEOUT = 8.0
CACHE_TTL = 21600
VALIDATE_TIMEOUT = 1.6
VALIDATE_RETRY_TIMEOUT = 4.0
UNKNOWN_CACHE_TTL = 60
UC_VALID_CACHE_TTL = 300
UC_SEARCH_CACHE_TTL = 300
BAIDU_VALID_CACHE_TTL = 300
BAIDU_SEARCH_CACHE_TTL = 300
MASTER_SYNC_VALIDATE_BUDGET = 2.5
BUDGET_UNKNOWN_TYPES = {'115', '189', '139', 'other'}
VALIDATE_WORKERS = 96
FAST_MAX_CHANNELS = 9
FULL_MAX_CHANNELS = 150
SEARCH_CACHE = {}
VALIDATE_CACHE = {}
VALIDATION_TASKS = {}
TASK_LOCK = threading.Lock()
RECENT_REQUESTS = []
REQUEST_LOG_LOCK = threading.Lock()
REQUEST_LOG_LIMIT = 200
TASK_TTL = 3600
EXCLUDED_CHANNELS = {'yoyokuakeduanju'}
PRIORITY_CHANNELS = [
    'Quark_Movies', 'leoziyuan', 'yunpanxunlei', 'NewQuark', 'shareAliyun',
    'alyp_TV', 'alyp_4K_Movies', 'ucquark', 'kuakeyun', 'tianyifc',
    'yunpan189', 'yunpanuc'
]

CLOUD_DOMAINS = (
    'pan.quark.cn', 'drive.uc.cn', 'pan.baidu.com', 'alipan.com',
    'aliyundrive.com', '115.com', '115cdn.com', 'anxia.com',
    '123pan.com', '123pan.cn', '123684.com', '123865.com', '123912.com',
    '123592.com', 'cloud.189.cn', 'caiyun.139.com', 'pan.xunlei.com',
)
URL_RE = re.compile(r'https?://[^\s<>"\']+', re.I)
NON_VIDEO_TITLE_RE = re.compile(r'(?:有声剧|有声书|听书|广播剧|多人有声|主播[：:]|音频|评书|小说朗读|朗读版)')


def is_cloud_url(url):
    try:
        host = urllib.parse.urlparse(url).netloc.lower()
    except Exception:
        return False
    return any(d in host for d in CLOUD_DOMAINS)




def is_excluded_cloud_url(url):
    try:
        host = urllib.parse.urlparse(url).netloc.lower()
    except Exception:
        return False
    return 'alipan.com' in host or 'aliyundrive.com' in host

def clean_tianyi_url(url):
    raw = (url or '').strip().replace('&amp;', '&')
    decoded = urllib.parse.unquote(raw)
    m = re.search(r'(?:[?&]code=|/web/share\?code=)([A-Za-z0-9]+)', decoded)
    if not m:
        return raw
    code = m.group(1)
    access = ''
    q = urllib.parse.parse_qs(urllib.parse.urlparse(raw).query)
    for key in ('accessCode', 'access_code', 'pwd', 'password'):
        if q.get(key):
            access = (q.get(key) or [''])[0]
            break
    if not access:
        am = re.search(r'(?:访问码|提取码|密码)\s*[：:]\s*([A-Za-z0-9]{3,12})', decoded)
        if am:
            access = am.group(1)
    clean = 'https://cloud.189.cn/web/share?code=' + urllib.parse.quote(code)
    if access:
        clean += '&accessCode=' + urllib.parse.quote(access)
    return clean


def clean_url(url):
    url = (url or '').strip().replace('&amp;', '&').lstrip('#')
    if 'cloud.189.cn' in url:
        url = clean_tianyi_url(url)
    # Cloud share URLs do not need URL fragments.  Keeping fragments can break
    # the legacy result format because channel items are separated by "##".
    if '#' in url:
        url = url.split('#', 1)[0]
    return url.rstrip('。，,；;、)）]】>&?#')


def is_non_video_title(title):
    return bool(NON_VIDEO_TITLE_RE.search(title or ''))


def title_from_text(text, keyword):
    lines = [x.strip() for x in text.splitlines() if x.strip()]
    for i, line in enumerate(lines):
        m = re.match(r'^(?:名称|片名|标题)\s*[:：]\s*(.+)$', line)
        if m and m.group(1).strip():
            return m.group(1).strip()[:120]
        if re.match(r'^(?:名称|片名|标题)\s*[:：]?$', line) and i + 1 < len(lines):
            return lines[i + 1].strip()[:120]
    for line in lines:
        if keyword and keyword in line:
            return line[:120]
    return (lines[0][:120] if lines else keyword or '未知')


TITLE_LABEL_RE = re.compile(r'^(?:名称|片名|标题|剧名|资源名称)\s*[:：]\s*(.+)$')
TITLE_LABEL_ONLY_RE = re.compile(r'^(?:名称|片名|标题|剧名|资源名称)\s*[:：]?$')
DESC_LINE_RE = re.compile(r'^(?:简介|描述|介绍|剧情|资源简介|主演|导演|类型|地区|语言|年份|集数|更新|标签)\s*[:：]?', re.I)
PLATFORM_LABEL_RE = re.compile(r'^(?:UC|UC网盘|夸克|夸克网盘|百度|百度网盘|迅雷|迅雷云盘|P?123|123网盘|115|115网盘|天翼|天翼云|移动|移动云|和彩云|阿里|阿里云盘|网盘|云盘|链接|下载|备用|在线)\s*[:：：-]?.*$', re.I)


def is_probable_title_line(line):
    line = (line or '').strip()
    if not line or URL_RE.search(line) or line.startswith(('http://', 'https://')):
        return False
    if DESC_LINE_RE.match(line) or PLATFORM_LABEL_RE.match(line):
        return False
    # Long prose/synopsis is not a resource title. Exact mode should not keep a
    # link just because its description says e.g. "遮天三部曲...完美世界".
    if len(line) > 80:
        return False
    return True


def title_from_context(text, keyword):
    lines = [x.strip() for x in (text or '').splitlines() if x.strip()]
    if not lines:
        return ''
    for i, line in enumerate(lines):
        m = TITLE_LABEL_RE.match(line)
        if m and m.group(1).strip():
            return m.group(1).strip()[:120]
        if TITLE_LABEL_ONLY_RE.match(line) and i + 1 < len(lines):
            return lines[i + 1].strip()[:120]
    # Pick the nearest probable title line before the link, not synopsis lines.
    for line in reversed(lines):
        if is_probable_title_line(line):
            return line[:120]
    return ''




def norm_text_for_match(text):
    text = html_lib.unescape(text or '').lower()
    # Remove spaces/punctuation/decorative separators so variants like
    # "大丨主宰", "大｜主宰", "灵魂摆渡·十年" still match the plain keyword.
    text = re.sub(r'[丨｜|·・•\-—–_\s]+', '', text)
    return re.sub(r'[\W_]+', '', text, flags=re.U)


def title_candidates_from_text(text, keyword):
    lines = [x.strip() for x in (text or '').splitlines() if x.strip()]
    candidates = []
    for i, line in enumerate(lines):
        m = re.match(r'^(?:名称|片名|标题|剧名|资源名称)\s*[:：]\s*(.+)$', line)
        if m and m.group(1).strip():
            candidates.append(m.group(1).strip())
        if re.match(r'^(?:名称|片名|标题|剧名|资源名称)\s*[:：]?$', line) and i + 1 < len(lines):
            candidates.append(lines[i + 1].strip())
    # Most Telegram resource posts put the real title in the first few text
    # lines.  Keep these as candidates, but do not scan arbitrary long body text
    # in exact mode, otherwise unrelated posts mentioning the keyword are kept.
    for line in lines[:6]:
        if not URL_RE.search(line) and not line.startswith(('http://', 'https://')):
            candidates.append(line)
    title = title_from_text(text, keyword)
    if title:
        candidates.append(title)
    return list(dict.fromkeys(candidates))


def message_matches_keyword(text, keyword):
    if not keyword:
        return True
    return norm_text_for_match(keyword) in norm_text_for_match(text)


def context_has_probable_title(text):
    lines = [x.strip() for x in (text or '').splitlines() if x.strip()]
    for i, line in enumerate(lines):
        if TITLE_LABEL_RE.match(line) or TITLE_LABEL_ONLY_RE.match(line):
            return True
    return any(is_probable_title_line(line) for line in lines)


def keyword_title_line_from_context(text, keyword):
    kw = norm_text_for_match(keyword)
    if not kw:
        return ''
    lines = [x.strip() for x in (text or '').splitlines() if x.strip()]
    # Use only short title-like lines near the link.  This is intentionally not
    # a full-message fuzzy match, so synopsis lines such as "遮天三部曲...完美世界"
    # will not pull unrelated resources in.
    for line in reversed(lines):
        if is_probable_title_line(line) and kw in norm_text_for_match(line):
            return line[:120]
    return ''


def context_matches_exact_keyword(text, keyword):
    if not keyword:
        return True
    kw = norm_text_for_match(keyword)
    if not kw:
        return True
    title = title_from_context(text, keyword)
    return kw in norm_text_for_match(title)


def exact_title_matches_keyword(text, keyword):
    if not keyword:
        return True
    kw = norm_text_for_match(keyword)
    if not kw:
        return True
    for cand in title_candidates_from_text(text, keyword):
        if kw in norm_text_for_match(cand):
            return True
    return False


def plain_text_from_html(block):
    def repl_anchor(m):
        href = html_lib.unescape(m.group(1) or '')
        inner = m.group(2) or ''
        inner = re.sub(r'<br\s*/?>', '\n', inner, flags=re.I)
        inner = re.sub(r'<[^>]+>', '', inner)
        inner = html_lib.unescape(inner).strip()
        # Telegram often stores cloud URLs in href attributes while the visible
        # text is just "查看链接"/domain. Inject the href into plain text so the
        # per-link context filter can evaluate this link instead of falling back
        # to the whole message.
        if href:
            return (inner + '\n' if inner else '') + href
        return inner
    block = re.sub(r'<a[^>]+href=["\'](https?://[^"\']+)["\'][^>]*>(.*?)</a>', repl_anchor, block, flags=re.I | re.S)
    plain = re.sub(r'<br\s*/?>', '\n', block, flags=re.I)
    plain = re.sub(r'<[^>]+>', '', plain)
    return html_lib.unescape(plain)


def split_plain_by_links(plain):
    matches = list(URL_RE.finditer(plain or ''))
    if not matches:
        return []
    out = []
    prev_end = 0
    prev_context_start = 0
    for idx, m in enumerate(matches):
        # Context for a link starts after the previous link. This ties each link
        # to its closest local title/description instead of the whole message.
        raw_link = m.group(0)
        context = plain[prev_context_start:m.start()]
        # If there are many lines since the previous link, keep only the nearby
        # block to avoid an earlier matched title leaking into this link.
        lines = [x for x in context.splitlines()]
        if len(lines) > 8:
            context = '\n'.join(lines[-8:])
        out.append((raw_link, context))
        prev_context_start = m.end()
        prev_end = m.end()
    return out

def fetch_tg(channel, keyword):
    q = urllib.parse.quote(keyword)
    url = f'https://telegram.me/s/{channel}?q={q}'
    req = urllib.request.Request(url, headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    })
    with OPENER.open(req, timeout=FETCH_TIMEOUT) as resp:
        return resp.read().decode('utf-8', 'ignore')


def parse_channel(channel, keyword, exact=True):
    try:
        text = fetch_tg(channel, keyword)
    except Exception:
        return ''

    items, seen = [], set()
    blocks = re.findall(r'<div class="tgme_widget_message_text[^>]*>(.*?)</div>', text, re.S)
    if not blocks:
        blocks = [text]
    for block in blocks:
        plain = plain_text_from_html(block)
        if exact:
            # Coarse guard first, then strict per-link context filtering below.
            if not exact_title_matches_keyword(plain, keyword) and not message_matches_keyword(plain, keyword):
                continue
            link_contexts = split_plain_by_links(plain)
            # Some Telegram messages expose links only in href attributes; fall
            # back to the whole-message title only when no plain URL exists.
            if link_contexts:
                inherited_title = ''
                inherited_match = False
                for raw, context in link_contexts:
                    link = clean_url(html_lib.unescape(raw))
                    if not is_cloud_url(link) or is_excluded_cloud_url(link) or link in seen:
                        continue
                    local_title = title_from_context(context, keyword)
                    smart_title = keyword_title_line_from_context(context, keyword)
                    if smart_title and not local_title:
                        local_title = smart_title
                    local_has_title = bool(local_title) and context_has_probable_title(context)
                    local_match = bool(local_title) and (norm_text_for_match(keyword) in norm_text_for_match(local_title))
                    if local_has_title:
                        inherited_title = local_title
                        inherited_match = local_match
                    if local_match:
                        title = local_title
                    elif inherited_match and not local_has_title:
                        # Multi-cloud posts often list several links under the
                        # same title.  Later links may only have platform text
                        # like "UC:"/"迅雷:" in their local context; inherit the
                        # last matched title for those links.
                        title = inherited_title
                    else:
                        continue
                    seen.add(link)
                    if is_non_video_title(title):
                        continue
                    safe_title = title.replace('$$', ' ').replace('##', ' ').strip() or keyword or '未知'
                    items.append(f'{link}$${safe_title}')
                continue
            if not exact_title_matches_keyword(plain, keyword):
                continue
        elif not message_matches_keyword(plain, keyword):
            continue
        title = title_from_text(plain, keyword)
        if is_non_video_title(title):
            continue
        candidates = re.findall(r'href="(https?://[^"]+)"', block)
        candidates.extend(URL_RE.findall(plain))
        for raw in candidates:
            link = clean_url(html_lib.unescape(raw))
            if not is_cloud_url(link) or is_excluded_cloud_url(link) or link in seen:
                continue
            seen.add(link)
            safe_title = title.replace('$$', ' ').replace('##', ' ').strip() or keyword or '未知'
            items.append(f'{link}$${safe_title}')
    if not items:
        return ''
    return channel + '$$$' + '##'.join(items)


def normalize_channels(raw):
    channels = []
    excluded = {x.lower() for x in EXCLUDED_CHANNELS}
    for part in re.split(r'[,\n\r]+', raw or ''):
        part = part.strip().lstrip('@')
        if not part:
            continue
        part = re.sub(r'^https?://t\.me/(?:s/)?', '', part).split('/')[0].split('?')[0]
        if part.lower() in excluded:
            continue
        if re.match(r'^[A-Za-z0-9_]{3,64}$', part):
            channels.append(part)
    return list(dict.fromkeys(channels))


def cloud_type(url):
    host = urllib.parse.urlparse(url).netloc.lower()
    if 'pan.quark.cn' in host:
        return 'quark'
    if 'drive.uc.cn' in host:
        return 'uc'
    if 'pan.baidu.com' in host:
        return 'baidu'
    if 'alipan.com' in host or 'aliyundrive.com' in host:
        return 'alipan'
    if '115.com' in host or '115cdn.com' in host or 'anxia.com' in host:
        return '115'
    if '123pan' in host or re.search(r'123(684|865|912|592)\.com', host):
        return '123pan'
    if 'cloud.189.cn' in host:
        return '189'
    if 'caiyun.139.com' in host:
        return '139'
    if 'pan.xunlei.com' in host:
        return 'xunlei'
    return 'other'



XUNLEI_CLIENT_ID = "Xqp0kJBXWhwaTpB6"
XUNLEI_CLIENT_VERSION = "1.92.63"
XUNLEI_HOST = "pan.xunlei.com"
XUNLEI_SIGN_SALTS = (
    "y/Yo0rvIoedqTLS0qrHXFpCW4Df8/swwD46QhBUT/0CsU",
    "DtzqOSvQcC7RFUMElw",
    "v9Ot2jYbH8BKxnLKzN5Q/URMhp5vv2xmU5CgvPVPVl",
    "UME2ypk6J12sqt0jektN6QYD",
    "c2xXtD9YLc4WV56",
    "T",
    "Pp",
    "hNCZhqhxzlA+BkGyoPFizQAtMAQWxhPX8Jmhnf0MqNrkv99curO",
    "uRdzLQQCHmCVIwwGGF6h+FIp/gbE2yKxGlFpPEpKBYRV3E7NLP4DW",
)
XUNLEI_INVALID_PHRASES = (
    "分享已失效", "分享地址已失效", "分享不存在", "链接不存在", "文件不存在", "资源不存在",
    "分享的文件已经被取消", "分享已被取消", "分享被取消", "取消了分享", "已取消分享",
    "来晚了", "啊哦", "已过期", "链接已过期", "已被删除", "不存在或已删除",
    "页面不存在", "not found", "share not found", "expired", "deleted", "invalid share",
    "好友已取消了分享", "该分享已取消", "分享文件已被删除", "文件已被删除",
    "抱歉，该分享已被作者删除", "该分享已被作者删除", "分享已删除",
)


def extract_xunlei_share_id(url):
    try:
        path = urllib.parse.urlparse(url).path or ""
    except Exception:
        return ""
    m = re.search(r"/s/([A-Za-z0-9_-]+)", path)
    return m.group(1) if m else ""


def phrase_hit_xunlei(text, phrases=XUNLEI_INVALID_PHRASES):
    t = str(text or "").lower()
    for p in phrases:
        if str(p).lower() in t:
            return p
    return ""


def xunlei_md5(s):
    return hashlib.md5(str(s).encode("utf-8")).hexdigest()


def xunlei_captcha_sign(device_id, ts):
    h = XUNLEI_CLIENT_ID + XUNLEI_CLIENT_VERSION + XUNLEI_HOST + device_id + str(ts)
    for salt in XUNLEI_SIGN_SALTS:
        h = xunlei_md5(h + salt)
    return "1." + h


def xunlei_captcha_token(device_id):
    ts = str(int(time.time() * 1000))
    meta = {
        "username": "",
        "phone_number": "",
        "email": "",
        "package_name": XUNLEI_HOST,
        "client_version": XUNLEI_CLIENT_VERSION,
        "captcha_sign": xunlei_captcha_sign(device_id, ts),
        "timestamp": ts,
        "user_id": "0",
    }
    body = {
        "client_id": XUNLEI_CLIENT_ID,
        "action": "get:/drive/v1/share",
        "device_id": device_id,
        "meta": meta,
    }
    headers = {
        "Content-Type": "application/json",
        "Origin": "https://pan.xunlei.com",
        "Referer": "https://pan.xunlei.com/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36",
    }
    try:
        req = urllib.request.Request(
            "https://xluser-ssl.xunlei.com/v1/shield/captcha/init",
            data=json.dumps(body).encode("utf-8"),
            headers=headers,
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=VALIDATE_TIMEOUT) as resp:
            j = json.loads(resp.read().decode("utf-8", "ignore") or "{}")
    except Exception as e:
        return "", "captcha_network_error_%s" % type(e).__name__
    token = j.get("captcha_token") or ""
    if token:
        return token, ""
    return "", str(j.get("error_description") or j.get("error") or j)[:160]


def validate_xunlei_api(share_id, pass_code):
    typ = "xunlei"
    device_id = uuid.uuid4().hex
    token, err = xunlei_captcha_token(device_id)
    if not token:
        return make_validation("unknown", None, typ, "captcha_" + err)
    params = {
        "share_id": share_id,
        "pass_code": pass_code or "",
        "limit": "20",
        "pass_code_token": "",
        "page_token": "",
        "order": "DEFAULT_ORDER",
        "thumbnail_size": "SIZE_MEDIUM",
    }
    api = "https://api-pan.xunlei.com/drive/v1/share?" + urllib.parse.urlencode(params)
    headers = {
        "Accept": "application/json",
        "Origin": "https://pan.xunlei.com",
        "Referer": "https://pan.xunlei.com/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36",
        "x-device-id": device_id,
        "x-client-id": XUNLEI_CLIENT_ID,
        "x-captcha-token": token,
    }
    try:
        req = urllib.request.Request(api, headers=headers, method="GET")
        with urllib.request.urlopen(req, timeout=VALIDATE_TIMEOUT) as resp:
            text = resp.read().decode("utf-8", "ignore")
            status_code = resp.status
    except urllib.error.HTTPError as e:
        try:
            text = e.read().decode("utf-8", "ignore")
        except Exception:
            text = ""
        status_code = e.code
        # captcha/auth failures stay unknown
        if status_code in (400, 401, 403):
            try:
                j = json.loads(text or "{}")
                return make_validation("unknown", None, typ, "api_http_%s_%s" % (status_code, j.get("error") or ""))
            except Exception:
                return make_validation("unknown", None, typ, "api_http_%s" % status_code)
        if status_code in (404, 410, 451):
            return make_validation("invalid", False, typ, "api_http_%s" % status_code)
        return make_validation("unknown", None, typ, "api_http_%s" % status_code)
    except Exception as e:
        return make_validation("unknown", None, typ, "api_network_%s" % type(e).__name__)
    try:
        j = json.loads(text or "{}")
    except Exception:
        bad = phrase_hit_xunlei(text)
        if bad:
            return make_validation("invalid", False, typ, "api_phrase_" + bad)
        return make_validation("unknown", None, typ, "api_json_error")
    if j.get("error"):
        bad = phrase_hit_xunlei(str(j.get("error_description") or j.get("error") or ""))
        if bad:
            return make_validation("invalid", False, typ, "api_" + bad)
        return make_validation("unknown", None, typ, "api_" + str(j.get("error") or j.get("error_code"))[:80])
    status = str(j.get("share_status") or "").upper()
    status_text = str(j.get("share_status_text") or "")
    if status in ("DELETED", "EXPIRED", "NOT_FOUND", "SENSITIVE", "SENSITIVE_WORD", "SENSITIVE_RESOURCE"):
        return make_validation("invalid", False, typ, "api_status_" + status.lower())
    if phrase_hit_xunlei(status_text):
        return make_validation("invalid", False, typ, "api_" + status_text)
    files = j.get("files")
    try:
        file_num = int(j.get("file_num") or 0)
    except Exception:
        file_num = 0
    if status == "OK" and ((isinstance(files, list) and len(files) > 0) or file_num > 0):
        return make_validation("valid", True, typ, "api_ok_files")
    if status == "OK":
        # empty OK may be password-needed root or empty folder; keep unknown
        return make_validation("unknown", None, typ, "api_ok_empty")
    return make_validation("unknown", None, typ, "api_status_" + (status.lower() or "missing"))


def validate_xunlei_page(url):
    headers = {
        "Referer": "https://www.google.com/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36",
    }
    try:
        req = urllib.request.Request(url, headers=headers, method="GET")
        with urllib.request.urlopen(req, timeout=VALIDATE_TIMEOUT) as resp:
            text = resp.read(131072).decode("utf-8", "ignore")
            code = resp.status
    except urllib.error.HTTPError as e:
        try:
            text = e.read(65536).decode("utf-8", "ignore")
        except Exception:
            text = ""
        code = e.code
        if code in (404, 410, 451):
            return make_validation("invalid", False, "xunlei", "http_%s" % code)
        return make_validation("unknown", None, "xunlei", "http_%s" % code)
    except Exception as e:
        return make_validation("unknown", None, "xunlei", "network_%s" % type(e).__name__)
    bad = phrase_hit_xunlei(text)
    if code in (404, 410, 451):
        return make_validation("invalid", False, "xunlei", "http_%s" % code)
    if bad:
        return make_validation("invalid", False, "xunlei", "phrase_" + bad)
    # SPA shell often returns 200 with "下载"; never treat that as valid.
    return make_validation("unknown", None, "xunlei", "page_http_%s" % code)


def validate_xunlei(url):
    typ = "xunlei"
    sid = extract_xunlei_share_id(url)
    if not sid:
        return make_validation("unknown", None, typ, "no_share_id")
    q = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
    pass_code = (q.get("pwd") or q.get("pass_code") or q.get("passcode") or q.get("password") or [""])[0]
    api_res = validate_xunlei_api(sid, pass_code)
    if api_res.get("status") in ("valid", "invalid"):
        return api_res
    page_res = validate_xunlei_page(url)
    if page_res.get("status") == "invalid":
        return page_res
    return api_res if api_res.get("status") == "unknown" else page_res


def make_validation(status='unknown', ok=None, typ='other', reason='', http_status=None, final_url=None, content_type=''):
    info = {'status': status, 'ok': ok, 'http_status': http_status, 'type': typ}
    if reason:
        info['reason'] = reason
    if final_url:
        info['final_url'] = final_url
    if content_type:
        info['content_type'] = content_type
    return info


def open_json(req, timeout=None):
    try:
        with urllib.request.urlopen(req, timeout=timeout or VALIDATE_TIMEOUT) as resp:
            body = resp.read(200000)
            return resp.status, resp.geturl(), resp.headers.get('Content-Type') or '', json.loads(body.decode('utf-8', 'ignore') or '{}')
    except urllib.error.HTTPError as e:
        body = e.read(200000)
        text = body.decode('utf-8', 'ignore')
        try:
            data = json.loads(text or '{}')
        except Exception:
            data = {'_raw': text[:1000]}
        return e.code, e.geturl(), (e.headers.get('Content-Type') if e.headers else '') or '', data


def validate_quark(url, timeout=None):
    typ = 'quark'
    u = urllib.parse.urlparse(url)
    m = re.search(r'/s/([A-Za-z0-9]+)', u.path)
    if not m:
        return make_validation('invalid', False, typ, 'bad_format')
    pwd_id = m.group(1)
    passcode = urllib.parse.parse_qs(u.query).get('pwd', [''])[0]
    token_api = 'https://drive-h.quark.cn/1/clouddrive/share/sharepage/token'
    data = json.dumps({'pwd_id': pwd_id, 'passcode': passcode, 'support_visit_limit_private_share': True}).encode()
    req = urllib.request.Request(token_api, data=data, method='POST', headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Accept': 'application/json, text/plain, */*',
        'Accept-Language': 'zh-CN,zh;q=0.9',
        'Content-Type': 'application/json', 'Origin': 'https://pan.quark.cn', 'Referer': 'https://pan.quark.cn/'
    })
    try:
        status, final_url, ct, res = open_json(req, timeout=timeout)
    except Exception as e:
        return make_validation('unknown', None, typ, type(e).__name__)
    code = res.get('code')
    if code == 41008:
        return make_validation('valid' if not passcode else 'invalid', True if not passcode else False, typ, 'password_required' if not passcode else 'bad_password', status, url, ct)
    if code in (41004, 41010, 41011):
        return make_validation('invalid', False, typ, 'bad_password', status, url, ct)
    if not (status == 200 and code == 0 and res.get('data', {}).get('stoken')):
        return make_validation('invalid', False, typ, f'code={code}', status, url, ct)
    stoken = res.get('data', {}).get('stoken')
    # PanCheck-style second phase: verify file list and share status.
    detail = 'https://drive-pc.quark.cn/1/clouddrive/share/sharepage/detail?' + urllib.parse.urlencode({'pwd_id': pwd_id, 'stoken': stoken, 'ver': '2', 'pr': 'ucpro'})
    dreq = urllib.request.Request(detail, headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Accept': 'application/json, text/plain, */*',
        'Accept-Language': 'zh-CN,zh;q=0.9',
        'Origin': 'https://pan.quark.cn', 'Referer': 'https://pan.quark.cn/',
        'Cache-Control': 'no-cache', 'Pragma': 'no-cache'
    })
    try:
        dstatus, dfinal, dct, detail_res = open_json(dreq, timeout=timeout)
    except Exception as e:
        # Token is valid, but detail failed due network/rate-limit: do not mark invalid.
        return make_validation('valid', True, typ, 'token_ok_detail_' + type(e).__name__, status, url, ct)
    data = detail_res.get('data') or {}
    lst = data.get('list') or []
    share = data.get('share') or {}
    share_status = share.get('status')
    partial = bool(share.get('partial_violation'))
    if not lst:
        if isinstance(share_status, int) and share_status > 1:
            return make_validation('invalid', False, typ, f'share_status={share_status}', dstatus, dfinal, dct)
        return make_validation('invalid', False, typ, 'empty_file_list', dstatus, dfinal, dct)
    if share_status == 1:
        return make_validation('valid', True, typ, 'detail_ok_partial_violation' if partial else 'detail_ok', dstatus, dfinal, dct)
    if share_status == 3 and not partial:
        return make_validation('valid', True, typ, 'share_status_3_no_violation', dstatus, dfinal, dct)
    if isinstance(share_status, int) and share_status > 1:
        return make_validation('invalid', False, typ, f'share_status={share_status}', dstatus, dfinal, dct)
    return make_validation('valid', True, typ, 'detail_ok', dstatus, dfinal, dct)

def validate_uc(url):
    typ = 'uc'
    m = re.search(r'/s/([A-Za-z0-9]+)', urllib.parse.urlparse(url).path)
    if not m:
        return make_validation('invalid', False, typ, 'bad_format')
    pwd_id = m.group(1)
    passcode = urllib.parse.parse_qs(urllib.parse.urlparse(url).query).get('pwd', [''])[0]

    # UC share page is a generic SPA shell and can contain "文件/分享" even for
    # dead links. Use the same token API as the front-end. This reliably returns
    # code=41011/41012 for expired/cancelled shares.
    token_api = 'https://pc-api.uc.cn/1/clouddrive/share/sharepage/token'
    data = json.dumps({'pwd_id': pwd_id, 'passcode': passcode, 'support_visit_limit_private_share': True}).encode()
    req = urllib.request.Request(token_api, data=data, method='POST', headers={
        'User-Agent': 'Mozilla/5.0 (Linux; Android 10) AppleWebKit/537.36 Chrome/120 Mobile Safari/537.36',
        'Accept': 'application/json, text/plain, */*',
        'Accept-Language': 'zh-CN,zh;q=0.9',
        'Content-Type': 'application/json',
        'Origin': 'https://drive.uc.cn',
        'Referer': 'https://drive.uc.cn/s/' + pwd_id,
    })
    try:
        status, final_url, ct, res = open_json(req, timeout=VALIDATE_RETRY_TIMEOUT)
    except Exception as e:
        return make_validation('unknown', None, typ, 'token_' + type(e).__name__)
    code = res.get('code')
    msg = str(res.get('message') or '')
    if status == 200 and code == 0 and (res.get('data') or {}).get('stoken'):
        data_obj = res.get('data') or {}
        stoken = data_obj.get('stoken')
        file_num = data_obj.get('file_num')
        if isinstance(file_num, int) and file_num <= 0:
            return make_validation('invalid', False, typ, 'empty_file_num', status, final_url, ct)
        # Second phase: verify share status and actual file list. Token alone can
        # be OK while the share detail/list is unavailable or violation-blocked.
        detail_qs = urllib.parse.urlencode({
            'pwd_id': pwd_id,
            'stoken': stoken,
            'pdir_fid': '0',
            'force': '0',
            '_page': '1',
            '_size': '50',
            '_fetch_banner': '1',
            '_fetch_share': '1',
            '_fetch_total': '1',
            '_sort': '',
        })
        dreq = urllib.request.Request('https://pc-api.uc.cn/1/clouddrive/share/sharepage/detail?' + detail_qs, headers={
            'User-Agent': 'Mozilla/5.0 (Linux; Android 10) AppleWebKit/537.36 Chrome/120 Mobile Safari/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Origin': 'https://drive.uc.cn',
            'Referer': 'https://drive.uc.cn/s/' + pwd_id,
        })
        try:
            dstatus, dfinal, dct, detail_res = open_json(dreq, timeout=VALIDATE_RETRY_TIMEOUT)
        except Exception as e:
            # Token is reliable enough to avoid false invalid on a transient detail failure.
            return make_validation('valid', True, typ, 'token_ok_detail_' + type(e).__name__, status, final_url, ct)
        dcode = detail_res.get('code')
        dmsg = str(detail_res.get('message') or '')
        if dcode in (41011, 41012, 41013, 41014) or any(k in dmsg for k in ['失效', '不存在', '取消', '删除', '过期', '违规']):
            return make_validation('invalid', False, typ, f'detail_code={dcode}', dstatus, dfinal, dct)
        ddata = detail_res.get('data') or {}
        share = ddata.get('share') or {}
        lst = ddata.get('list') or []
        share_status = share.get('status')
        partial = bool(share.get('partial_violation'))
        all_file_num = share.get('all_file_num')
        file_only_num = share.get('file_only_num')
        if isinstance(share_status, int) and share_status > 1 and not partial:
            return make_validation('invalid', False, typ, f'share_status={share_status}', dstatus, dfinal, dct)
        if not lst and ((isinstance(all_file_num, int) and all_file_num <= 0) or (isinstance(file_only_num, int) and file_only_num <= 0)):
            return make_validation('invalid', False, typ, 'empty_detail_list', dstatus, dfinal, dct)
        if not lst and share_status not in (1, None):
            return make_validation('invalid', False, typ, 'empty_detail_list', dstatus, dfinal, dct)
        return make_validation('valid', True, typ, 'detail_ok_partial_violation' if partial else 'detail_ok', dstatus, dfinal, dct)
    if code in (41004, 41010):
        return make_validation('invalid', False, typ, 'bad_password', status, final_url, ct)
    if code in (41011, 41012, 41013, 41014) or any(k in msg for k in ['失效', '不存在', '取消', '删除', '过期', '违规']):
        return make_validation('invalid', False, typ, f'code={code}', status, final_url, ct)
    if code == 41008:
        return make_validation('valid' if not passcode else 'invalid', True if not passcode else False, typ, 'password_required' if not passcode else 'bad_password', status, final_url, ct)
    return make_validation('unknown', None, typ, f'code={code}', status, final_url, ct)


def baidu_ids(url):
    u = urllib.parse.urlparse(url)
    q = urllib.parse.parse_qs(u.query)
    pwd = (q.get('pwd') or [''])[0]
    if u.path.startswith('/s/'):
        surl = u.path.split('/s/', 1)[1].split('/')[0]
        short = surl[1:] if len(surl) > 1 and surl.startswith('1') else surl
        return short, pwd
    if u.path.startswith('/share/init') or u.path.startswith('/wap/init'):
        return (q.get('surl') or [''])[0], pwd
    return '', pwd


def validate_baidu(url):
    typ = 'baidu'
    short, pwd = baidu_ids(url)
    if not short:
        return make_validation('invalid', False, typ, 'bad_format')
    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36', 'Referer': url, 'Accept': 'application/json, text/plain, */*'}
    cookie = ''
    if pwd:
        verify = f'https://pan.baidu.com/share/verify?surl={urllib.parse.quote(short)}&pwd={urllib.parse.quote(pwd)}'
        body = urllib.parse.urlencode({'pwd': pwd, 'vcode': '', 'vcode_str': ''}).encode()
        req = urllib.request.Request(verify, data=body, method='POST', headers={**headers, 'Content-Type': 'application/x-www-form-urlencoded'})
        try:
            _, _, _, res = open_json(req)
            if res.get('errno') != 0:
                return make_validation('invalid', False, typ, f"verify_errno={res.get('errno')}")
            if res.get('randsk'):
                cookie = 'BDCLND=' + res.get('randsk')
        except Exception as e:
            return make_validation('unknown', None, typ, 'verify_' + type(e).__name__)
    api = f'https://pan.baidu.com/share/list?web=5&app_id=250528&desc=1&showempty=0&page=1&num=20&order=time&shorturl={urllib.parse.quote(short)}&root=1&view_mode=1&channel=chunlei&web=1&clienttype=0'
    h = dict(headers)
    if cookie:
        h['Cookie'] = cookie
    req = urllib.request.Request(api, headers=h)
    try:
        status, final_url, ct, res = open_json(req)
    except Exception as e:
        return make_validation('unknown', None, typ, type(e).__name__)
    errno = int(res.get('errno', 999999))
    if errno == 0:
        return make_validation('valid', True, typ, 'errno_0', status, final_url, ct)
    if errno == -12:
        return make_validation('valid', True, typ, 'password_required', status, final_url, ct)
    if errno == -62:
        return make_validation('unknown', None, typ, 'rate_limited', status, final_url, ct)
    return make_validation('invalid', False, typ, f'errno={errno}', status, final_url, ct)


def validate_alipan(url):
    typ = 'alipan'
    parts = [x for x in urllib.parse.urlparse(url).path.split('/') if x]
    if not parts:
        return make_validation('invalid', False, typ, 'bad_format')
    share_id = parts[-1]
    api = f'https://api.aliyundrive.com/adrive/v3/share_link/get_share_by_anonymous?share_id={urllib.parse.quote(share_id)}'
    data = json.dumps({'share_id': share_id}).encode()
    req = urllib.request.Request(api, data=data, method='POST', headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Content-Type': 'application/json', 'Origin': 'https://www.alipan.com', 'Referer': 'https://www.alipan.com/', 'x-canary': 'client=web,app=share,version=v2.3.1'
    })
    try:
        status, final_url, ct, res = open_json(req)
    except urllib.error.HTTPError as e:
        return make_validation('unknown' if e.code == 429 else 'invalid', None if e.code == 429 else False, typ, f'http_{e.code}', e.code)
    except Exception as e:
        return make_validation('unknown', None, typ, type(e).__name__)
    code = res.get('code') or ''
    if 'Exceed' in code:
        return make_validation('unknown', None, typ, 'rate_limited', status, final_url, ct)
    if 'ShareLink' in code:
        return make_validation('invalid', False, typ, code, status, final_url, ct)
    if res.get('file_count', 0) > 0 or res.get('share_name') or res.get('share_title'):
        return make_validation('valid', True, typ, 'api_ok', status, final_url, ct)
    return make_validation('unknown', None, typ, 'empty_response', status, final_url, ct)


def decode_123pan_uid(prefix):
    code62 = 'Tvd3hHA9QEkom14xpfaBJIMwgFYGPXn2sWCNORDr80KuUSl7bZcetizL5q6yVj'
    dec = {ch: i for i, ch in enumerate(code62)}
    if not prefix or any(ch not in dec for ch in prefix):
        return None
    value = 0
    for i, ch in enumerate(prefix):
        value += dec[ch] * (62 ** i)
    if value <= 0 or value > 9007199254740991:
        return None
    return value


def validate_123(url):
    typ = '123pan'
    parsed = urllib.parse.urlparse(url)
    path = parsed.path
    m = re.search(r'/(?:s|ps|123pan)/([A-Za-z0-9]+-[A-Za-z0-9]+)', path)
    if not m:
        return make_validation('invalid', False, typ, 'bad_format')
    share_key = m.group(1)
    q = urllib.parse.parse_qs(parsed.query)
    share_pwd = (q.get('pwd') or q.get('password') or q.get('SharePwd') or [''])[0]
    uid_prefix = share_key.split('-', 1)[0]
    uid = decode_123pan_uid(uid_prefix)
    if not uid:
        return make_validation('unknown', None, typ, 'bad_share_uid')
    base = 'https://%s.share.123pan.cn' % uid
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Accept': 'application/json, text/plain, */*',
        'Referer': '%s/123pan/%s?notoken=1' % (base, share_key),
        'Origin': base,
    }
    # 1) share metadata: cancelled/expired/missing shares usually fail here.
    info_api = '%s/gsb/s/%s' % (base, urllib.parse.quote(share_key))
    try:
        status, final_url, ct, res = open_json(urllib.request.Request(info_api, headers=headers), timeout=VALIDATE_RETRY_TIMEOUT)
    except Exception as e:
        return make_validation('unknown', None, typ, type(e).__name__)
    info = res.get('info') if isinstance(res.get('info'), dict) else res
    code = info.get('code')
    data = info.get('data') or {}
    msg = str(info.get('message') or info.get('msg') or '')
    if code in (5103, 5104, 5105, 5107, 5110, 5111) or any(x in msg for x in ['不存在', '失效', '删除', '取消', '过期']):
        return make_validation('invalid', False, typ, f'code={code}:{msg}', status, final_url, ct)
    if code not in (0, None):
        return make_validation('unknown', None, typ, f'code={code}:{msg}', status, final_url, ct)
    # 2) file list: some cancelled/cleaned shares still return share metadata
    # with ShareName, but list is empty ("共0项/暂无文件"). Treat empty list as invalid.
    list_api = '%s/gsb/s/share-list?%s' % (
        base,
        urllib.parse.urlencode({'OrderId': '', 'SharePwd': share_pwd or '', 'shareKey': share_key}),
    )
    try:
        lstatus, lfinal, lct, lres = open_json(urllib.request.Request(list_api, headers=headers), timeout=VALIDATE_RETRY_TIMEOUT)
    except Exception as e:
        # metadata exists, but list unavailable; keep unknown rather than false valid.
        return make_validation('unknown', None, typ, 'list_' + type(e).__name__, status, final_url, ct)
    lcode = lres.get('code')
    lmsg = str(lres.get('message') or lres.get('msg') or '')
    ldata = lres.get('data') or {}
    if lcode in (5103, 5104, 5105, 5107, 5110, 5111) or any(x in lmsg for x in ['不存在', '失效', '删除', '取消', '过期']):
        return make_validation('invalid', False, typ, f'list_code={lcode}:{lmsg}', lstatus, lfinal, lct)
    info_list = ldata.get('InfoList') if isinstance(ldata, dict) else None
    try:
        list_len = int(ldata.get('Len')) if isinstance(ldata, dict) and ldata.get('Len') is not None else None
    except Exception:
        list_len = None
    if isinstance(info_list, list):
        if len(info_list) == 0 or list_len == 0:
            return make_validation('invalid', False, typ, 'empty_file_list', lstatus, lfinal, lct)
        return make_validation('valid', True, typ, 'list_ok', lstatus, lfinal, lct)
    if code == 0 and (data.get('ShareName') or data.get('FileName') or data.get('ShareKey')):
        # Metadata ok but list shape unexpected.
        return make_validation('unknown', None, typ, 'list_shape_unknown', lstatus, lfinal, lct)
    return make_validation('unknown', None, typ, f'code={code}:{msg}', status, final_url, ct)

def validate_115(url):
    typ = '115'
    u = urllib.parse.urlparse(url)
    share_code = [x for x in u.path.split('/') if x]
    share_code = share_code[-1] if share_code else ''
    q = urllib.parse.parse_qs(u.query)
    receive = (q.get('password') or [''])[0]
    if not receive and u.fragment and 'password=' in u.fragment:
        receive = (urllib.parse.parse_qs(u.fragment).get('password') or [''])[0]
    if not share_code or not receive:
        return make_validation('unknown', None, typ, 'missing_password')
    api = f'https://115cdn.com/webapi/share/snap?share_code={urllib.parse.quote(share_code)}&offset=0&limit=20&receive_code={urllib.parse.quote(receive)}&cid='
    req = urllib.request.Request(api, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': f'https://115cdn.com/s/{share_code}?password={receive}&', 'X-Requested-With': 'XMLHttpRequest'})
    try:
        status, final_url, ct, res = open_json(req)
    except Exception as e:
        return make_validation('unknown', None, typ, type(e).__name__)
    if res.get('state') and res.get('errno') == 0:
        data = res.get('data') or {}
        share_state = data.get('share_state') or (data.get('shareinfo') or {}).get('share_state')
        if share_state == 1:
            return make_validation('valid', True, typ, 'api_ok', status, final_url, ct)
        return make_validation('invalid', False, typ, f'share_state={share_state}', status, final_url, ct)
    return make_validation('invalid', False, typ, res.get('error') or 'api_error', status, final_url, ct)


def tianyi_ids(url):
    u = urllib.parse.urlparse(clean_tianyi_url(url))
    q = urllib.parse.parse_qs(u.query)
    return (q.get('code') or [''])[0], (q.get('accessCode') or [''])[0]


def validate_189(url):
    typ = '189'
    share_code, access_code = tianyi_ids(url)
    if not share_code:
        return make_validation('invalid', False, typ, 'bad_format')
    api = 'https://cloud.189.cn/api/open/share/getShareInfoByCode.action?' + urllib.parse.urlencode({
        'shareCode': share_code,
        'accessCode': access_code,
    })
    req = urllib.request.Request(api, headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Accept': 'application/json, text/plain, */*',
        'Origin': 'https://cloud.189.cn',
        'Referer': 'https://cloud.189.cn/web/share?code=' + urllib.parse.quote(share_code),
    })
    try:
        status, final_url, ct, res = open_json(req, timeout=VALIDATE_RETRY_TIMEOUT)
        text = json.dumps(res, ensure_ascii=False)
    except urllib.error.HTTPError as e:
        status = e.code; final_url = e.geturl(); ct = e.headers.get('Content-Type') or ''
        try:
            text = e.read(4096).decode('utf-8', 'ignore')
        except Exception:
            text = ''
    except Exception as e:
        return make_validation('unknown', None, typ, type(e).__name__)
    bad_codes = ['ShareAuditNotPass', 'ShareNotFound', 'ShareExpired', 'ShareCancelled', 'InvalidAccessCode', 'ShareNotExist']
    if any(code in text for code in bad_codes) or any(k in text for k in ['share audit not pass', '分享不存在', '分享已失效', '访问码错误', '提取码错误', '已取消']):
        reason = 'api_invalid'
        m = re.search(r'<code>([^<]+)</code>|"errorCode"\s*:\s*"([^"]+)"', text)
        if m:
            reason = m.group(1) or m.group(2) or reason
        return make_validation('invalid', False, typ, reason, status, final_url, ct)
    if status and 200 <= status < 300:
        return make_validation('valid', True, typ, 'api_ok', status, final_url, ct)
    return make_validation('unknown', None, typ, 'api_status=' + str(status), status, final_url, ct)


def validate_fallback(url, typ):
    info = {'status': 'unknown', 'ok': None, 'http_status': None, 'type': typ}
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        'Range': 'bytes=0-65535',
    }
    bad_markers = ['分享已失效', '文件不存在', '链接不存在', '已被取消', '已过期', '来晚啦', 'share not found', '分享不存在', '资源不存在', '链接已失效', 'page not found', '404 not found', '不存在或已被删除']
    status = None; final_url = url; ct = ''; body = b''
    try:
        req = urllib.request.Request(url, headers=headers, method='HEAD')
        with urllib.request.urlopen(req, timeout=VALIDATE_TIMEOUT) as resp:
            status = resp.status; final_url = resp.geturl(); ct = resp.headers.get('Content-Type') or ''
    except urllib.error.HTTPError as e:
        status = e.code; final_url = e.geturl(); ct = e.headers.get('Content-Type') or ''
        try: body = e.read(65536)
        except Exception: body = b''
    except Exception:
        status = None
    # For fallback platforms, do a small GET to catch visible invalid pages.
    if status is None or (200 <= status < 400):
        try:
            req = urllib.request.Request(url, headers=headers)
            with urllib.request.urlopen(req, timeout=VALIDATE_TIMEOUT) as resp:
                status = resp.status; final_url = resp.geturl(); ct = resp.headers.get('Content-Type') or ''; body = resp.read(65536)
        except urllib.error.HTTPError as e:
            status = e.code; final_url = e.geturl(); ct = e.headers.get('Content-Type') or ''
            try: body = e.read(65536)
            except Exception: body = b''
        except Exception as e:
            if status is None:
                info.update({'status': 'unknown', 'ok': None, 'error': type(e).__name__})
                return info
    text = body.decode('utf-8', 'ignore').lower() if body else ''
    bad = status in (404, 410) or any(m.lower() in text for m in bad_markers)
    if bad:
        info.update({'status': 'invalid', 'ok': False})
    elif status is not None and 200 <= status < 400:
        info.update({'status': 'valid', 'ok': True})
    else:
        info.update({'status': 'unknown', 'ok': None})
    info.update({'http_status': status, 'final_url': final_url, 'content_type': ct})
    return info

def validate_link(url, force=False, retry_timeout=None):
    now = time.time()
    cached = VALIDATE_CACHE.get(url)
    if cached and not force:
        cached_t, cached_info = cached
        if (cached_info or {}).get('status') == 'unknown':
            ttl = UNKNOWN_CACHE_TTL
        elif cloud_type(url) == 'uc' and (cached_info or {}).get('status') == 'valid':
            ttl = UC_VALID_CACHE_TTL
        elif cloud_type(url) == 'baidu' and (cached_info or {}).get('status') == 'valid':
            ttl = BAIDU_VALID_CACHE_TTL
        else:
            ttl = CACHE_TTL
        if now - cached_t < ttl:
            return cached_info
    typ = cloud_type(url)
    try:
        if typ == 'quark':
            info = validate_quark(url, timeout=retry_timeout)
        elif typ == 'uc':
            info = validate_uc(url)
        elif typ == 'baidu':
            info = validate_baidu(url)
        elif typ == 'alipan':
            info = validate_alipan(url)
        elif typ == '123pan':
            info = validate_123(url)
        elif typ == '115':
            info = validate_115(url)
        elif typ == '189':
            info = validate_189(url)
        elif typ == 'xunlei':
            info = validate_xunlei(url)
        else:
            info = validate_fallback(url, typ)
    except Exception as e:
        info = make_validation('unknown', None, typ, type(e).__name__)
    info['type'] = typ
    VALIDATE_CACHE[url] = (now, info)
    return info


def retry_unknown_validity(validity):
    # Unknown is kept to avoid false deletion.  Retry only high-value platforms
    # where a transient timeout often hides a clear invalid code.  Skip slower
    # low-value fallback platforms (189/139/115/other) to keep cold search fast.
    retry_types = {'quark', 'uc', 'baidu', '123pan'}
    unknown_links = [link for link, info in validity.items() if (info or {}).get('status') == 'unknown' and cloud_type(link) in retry_types]
    if not unknown_links:
        return validity
    with ThreadPoolExecutor(max_workers=min(4, len(unknown_links))) as pool:
        future_map = {pool.submit(validate_link, link, True, VALIDATE_RETRY_TIMEOUT): link for link in unknown_links}
        for fut in as_completed(future_map):
            link = future_map[fut]
            try:
                info = fut.result()
            except Exception as e:
                info = {'status': 'unknown', 'ok': None, 'type': cloud_type(link), 'reason': type(e).__name__}
            old = validity.get(link) or {}
            if info.get('status') != 'unknown' or old.get('status') == 'unknown':
                validity[link] = info
    return validity



def retry_xunlei_validity(validity):
    """Recheck every Xunlei link that is not already API-backed.

    Remote/old workers may mark deleted Xunlei SPA pages as valid (HTTP 200).
    Only trust reasons produced by validate_xunlei()/validate_xunlei_api().
    """
    if not isinstance(validity, dict):
        return validity
    for link, info in list(validity.items()):
        if cloud_type(link) != 'xunlei':
            continue
        reason = str((info or {}).get('reason') or '')
        if reason.startswith('api_status_') or reason in ('api_ok_files', 'api_ok_empty'):
            continue
        try:
            validity[link] = validate_xunlei(link)
        except Exception as e:
            validity[link] = make_validation('unknown', None, 'xunlei', 'master_recheck_%s' % type(e).__name__)
    return validity

    for link, info in list(validity.items()):
        if cloud_type(link) != 'xunlei':
            continue
        info = info or {}
        reason = str(info.get('reason') or '')
        # Already API-backed definitive result.
        if reason.startswith('api_status_') or reason in ('api_ok_files', 'api_ok_empty') or reason.startswith('api_http_') or reason.startswith('api_phrase_') or reason.startswith('api_') and info.get('status') in ('valid', 'invalid') and 'captcha' not in reason:
            # still recheck false-valid page-style even if status valid without api reason
            if reason.startswith('api_') and info.get('status') in ('valid', 'invalid'):
                continue
        # Page/fallback style: has content_type html or bare http_status without api reason
        needs = (
            not reason.startswith('api_')
            or (info.get('status') == 'valid' and str(info.get('content_type') or '').startswith('text/html'))
            or (info.get('status') == 'valid' and info.get('http_status') == 200 and not reason.startswith('api_'))
        )
        if not needs and info.get('status') in ('valid', 'invalid') and reason.startswith('api_'):
            continue
        try:
            validity[link] = validate_xunlei(link)
        except Exception as e:
            validity[link] = make_validation('unknown', None, 'xunlei', 'master_recheck_%s' % type(e).__name__)
    return validity


def retry_unknown_123pan_validity(validity):
    unknown_links = [link for link, info in (validity or {}).items() if (info or {}).get('status') == 'unknown' and cloud_type(link) == '123pan']
    if not unknown_links:
        return validity
    with ThreadPoolExecutor(max_workers=min(8, len(unknown_links))) as pool:
        future_map = {pool.submit(validate_link, link, True, VALIDATE_RETRY_TIMEOUT): link for link in unknown_links}
        for fut in as_completed(future_map):
            link = future_map[fut]
            try:
                info = fut.result()
            except Exception as e:
                info = {'status': 'unknown', 'ok': None, 'type': '123pan', 'reason': type(e).__name__}
            if info.get('status') != 'unknown':
                validity[link] = info
    return validity


def extract_links_from_results(results):
    links = []
    for result in results:
        if not result or '$$$' not in result:
            continue
        _, content = result.split('$$$', 1)
        for item in content.split('##'):
            parts = item.split('$$', 1)
            if len(parts) >= 2:
                links.append(parts[0].split('@', 1)[0])
    return list(dict.fromkeys(links))


def search_cache_ttl(payload):
    # UC/Baidu links can flip from valid to invalid quickly, and each Space has
    # its own in-memory validation cache.  Keep final search responses containing
    # those links short-lived so stale valid shares are rechecked soon.
    try:
        links = extract_links_from_results((payload or {}).get('results') or [])
        types = {cloud_type(link) for link in links}
        if 'uc' in types:
            return UC_SEARCH_CACHE_TTL
        if 'baidu' in types:
            return BAIDU_SEARCH_CACHE_TTL
    except Exception:
        pass
    return CACHE_TTL


def filter_results_by_validity(results, validity):
    filtered = []
    for result in results:
        if not result or '$$$' not in result:
            filtered.append(result)
            continue
        ch, content = result.split('$$$', 1)
        kept = []
        for item in content.split('##'):
            parts = item.split('$$', 1)
            if len(parts) < 2:
                continue
            link = parts[0].split('@', 1)[0]
            if validity.get(link, {}).get('status') != 'invalid':
                kept.append(item)
        filtered.append(ch + '$$$' + '##'.join(kept) if kept else '')
    return filtered


def compact_results(results):
    return [r for r in (results or []) if r and '$$$' in r and r.split('$$$', 1)[1].strip()]


def validation_summary(validity, total_links):
    return {
        'checked': len(validity),
        'total_links': total_links,
        'valid': sum(1 for v in validity.values() if v.get('status') == 'valid'),
        'invalid': sum(1 for v in validity.values() if v.get('status') == 'invalid'),
        'unknown': sum(1 for v in validity.values() if v.get('status') == 'unknown'),
        'limited': False,
        'complete': len(validity) == total_links,
    }


def validate_links_sync(links):
    links = list(dict.fromkeys(links or []))
    validity = {}
    if links:
        pool = ThreadPoolExecutor(max_workers=min(VALIDATE_WORKERS, len(links)))
        future_map = {pool.submit(validate_link, link): link for link in links}
        for fut in as_completed(future_map):
            link = future_map[fut]
            try:
                validity[link] = fut.result()
            except Exception as e:
                validity[link] = {'status': 'unknown', 'ok': None, 'type': cloud_type(link), 'reason': type(e).__name__}
        pool.shutdown(wait=False, cancel_futures=True)
    retry_unknown_validity(validity)
    return validity, validation_summary(validity, len(links))


def validate_links_budgeted(links, budget_sec):
    links = list(dict.fromkeys(links or []))
    validity = {}
    if not links:
        return validity, validation_summary(validity, 0)
    budget_sec = max(0.2, float(budget_sec or MASTER_SYNC_VALIDATE_BUDGET))
    pool = ThreadPoolExecutor(max_workers=min(VALIDATE_WORKERS, len(links)))
    future_map = {pool.submit(validate_link, link): link for link in links}
    done, pending = wait(future_map.keys(), timeout=budget_sec)
    for fut in done:
        link = future_map[fut]
        try:
            validity[link] = fut.result()
        except Exception as e:
            validity[link] = {'status': 'unknown', 'ok': None, 'type': cloud_type(link), 'reason': type(e).__name__}

    low_timeout = 0
    strict_pending = []
    for fut in pending:
        link = future_map[fut]
        typ = cloud_type(link)
        if typ in BUDGET_UNKNOWN_TYPES:
            low_timeout += 1
            fut.cancel()
            validity[link] = {'status': 'unknown', 'ok': None, 'type': typ, 'reason': 'budget_timeout_low_priority'}
        else:
            strict_pending.append(fut)

    # Scheme 3 + 2.5s: low-priority fallback platforms may remain unknown,
    # but mainstream platforms keep waiting so they are not unknown only because
    # of the global budget.
    for fut in as_completed(strict_pending):
        link = future_map[fut]
        try:
            validity[link] = fut.result()
        except Exception as e:
            validity[link] = {'status': 'unknown', 'ok': None, 'type': cloud_type(link), 'reason': type(e).__name__}

    pool.shutdown(wait=False, cancel_futures=True)
    retry_unknown_validity(validity)
    summary = validation_summary(validity, len(links))
    summary['budgeted'] = True
    summary['budget_sec'] = budget_sec
    summary['timed_out'] = low_timeout
    summary['budget_unknown_types'] = sorted(BUDGET_UNKNOWN_TYPES)
    return validity, summary


def validate_results_sync(results, drop_invalid=True):
    links = extract_links_from_results(results)
    validity, summary = validate_links_sync(links)
    out_results = filter_results_by_validity(results, validity) if drop_invalid else results
    return out_results, validity, summary


def validate_task_worker(task_id):
    with TASK_LOCK:
        task = VALIDATION_TASKS.get(task_id)
        if not task:
            return
        task['status'] = 'running'
        task['started_at'] = time.time()
    links = task.get('links') or []
    validity = {}
    if links:
        pool = ThreadPoolExecutor(max_workers=min(VALIDATE_WORKERS, len(links)))
        future_map = {pool.submit(validate_link, link): link for link in links}
        for fut in as_completed(future_map):
            link = future_map[fut]
            try:
                info = fut.result()
            except Exception as e:
                info = {'status': 'unknown', 'ok': None, 'error': type(e).__name__}
            validity[link] = info
            with TASK_LOCK:
                t = VALIDATION_TASKS.get(task_id)
                if t:
                    t['validity'] = dict(validity)
                    t['validation'] = validation_summary(validity, len(links))
        pool.shutdown(wait=False, cancel_futures=True)
    retry_unknown_validity(validity)
    filtered = filter_results_by_validity(task.get('results') or [], validity)
    with TASK_LOCK:
        t = VALIDATION_TASKS.get(task_id)
        if t:
            t['validity'] = validity
            t['validation'] = validation_summary(validity, len(links))
            t['results'] = filtered
            t['non_empty_channels'] = sum(bool(x) for x in filtered)
            t['status'] = 'complete'
            t['completed_at'] = time.time()


def start_validation_task(results):
    links = extract_links_from_results(results)
    task_id = uuid.uuid4().hex[:16]
    task = {
        'task_id': task_id,
        'status': 'processing',
        'created_at': time.time(),
        'results': results,
        'links': links,
        'validity': {},
        'validation': validation_summary({}, len(links)),
        'non_empty_channels': sum(bool(x) for x in results),
    }
    with TASK_LOCK:
        now = time.time()
        for tid, old_task in list(VALIDATION_TASKS.items()):
            if now - old_task.get('created_at', now) > TASK_TTL:
                VALIDATION_TASKS.pop(tid, None)
        VALIDATION_TASKS[task_id] = task
    th = threading.Thread(target=validate_task_worker, args=(task_id,), daemon=True)
    th.start()
    return task_id, task


def task_response(task, include_validity=False):
    validation = task.get('validation') or validation_summary({}, len(task.get('links') or []))
    payload = {
        'task_id': task.get('task_id'),
        'status': task.get('status'),
        'validation_status': 'complete' if task.get('status') == 'complete' else 'processing',
        'validation': validation,
        'non_empty_channels': task.get('non_empty_channels'),
        'results': compact_results(task.get('results') or []),
    }
    if include_validity:
        payload['validity'] = task.get('validity') or {}
        payload['validity_count'] = len(payload['validity'])
    return payload


def handle_search_local(query):
    keyword = (query.get('keyword') or [''])[0].strip()
    channel_raw = (query.get('channelUsername') or query.get('channels') or [''])[0]
    channels = normalize_channels(channel_raw)
    requested_count = len(channels)
    mode_param = (query.get('mode') or query.get('search_mode') or ['full'])[0].lower()
    full_requested = mode_param not in ('fast', 'quick')
    priority = [ch for ch in PRIORITY_CHANNELS if ch in channels]
    rest = [ch for ch in channels if ch not in priority]
    channels = priority + rest
    if full_requested:
        channels = channels[:FULL_MAX_CHANNELS]
        max_workers = min(FULL_WORKERS, len(channels))
        search_timeout = FULL_SEARCH_TIMEOUT
        mode_name = 'full'
    else:
        channels = channels[:FAST_MAX_CHANNELS]
        max_workers = min(FAST_WORKERS, len(channels))
        search_timeout = FAST_SEARCH_TIMEOUT
        mode_name = 'fast'
    if not keyword or not channels:
        return 400, {'error': "Missing 'keyword' or 'channelUsername' parameter"}
    async_validate = first_value(query, 'async_validate', 'false').lower() in ('1', 'true', 'yes', 'on')
    validate_enabled = first_value(query, 'validate', 'true').lower() not in ('0', 'false', 'no', 'off', 'skip')
    exact_match = first_value(query, 'exact', 'true').lower() not in ('0', 'false', 'no', 'off', 'fuzzy')
    validate_mode = 'no_validate' if not validate_enabled else ('async_validate' if async_validate else 'sync_validate')
    exact_mode = 'exact' if exact_match else 'fuzzy'
    cache_key = (keyword, ','.join(channels), mode_name, validate_mode + '_' + exact_mode + '_smartlocal_diskshort_tianyi_v23')
    now = time.time()
    cached = SEARCH_CACHE.get(cache_key)
    if cached and now - cached[0] < search_cache_ttl(cached[1]):
        return 200, cached[1]

    results = [''] * len(channels)
    pool = ThreadPoolExecutor(max_workers=max_workers)
    future_map = {pool.submit(parse_channel, ch, keyword, exact_match): i for i, ch in enumerate(channels)}
    done, pending = wait(future_map.keys(), timeout=search_timeout)
    for fut in done:
        idx = future_map[fut]
        try:
            results[idx] = fut.result() or ''
        except Exception:
            results[idx] = ''
    for fut in pending:
        fut.cancel()
    pool.shutdown(wait=False, cancel_futures=True)

    if not validate_enabled:
        links = extract_links_from_results(results)
        visible_results = compact_results(results)
        payload = {
            'results': visible_results,
            'cached': False,
            'mode': mode_name,
            'searched_channels': len(channels),
            'requested_channels': requested_count,
            'timed_out_channels': len(pending),
            'non_empty_channels': len(visible_results),
            'validation_status': 'skipped',
            'validation': {**validation_summary({}, len(links)), 'skipped': True},
        }
    elif async_validate:
        task_id, task = start_validation_task(results)
        visible_results = compact_results(results)
        payload = {
            'results': visible_results,
            'cached': False,
            'mode': mode_name,
            'searched_channels': len(channels),
            'requested_channels': requested_count,
            'timed_out_channels': len(pending),
            'non_empty_channels': len(visible_results),
            'validation_status': 'processing',
            'task_id': task_id,
            'validation': {**task.get('validation', {}), 'task_id': task_id, 'poll_url': '/api/validation/' + task_id},
        }
    else:
        results, validity, validation = validate_results_sync(results, drop_invalid=True)
        visible_results = compact_results(results)
        payload = {
            'results': visible_results,
            'cached': False,
            'mode': mode_name,
            'searched_channels': len(channels),
            'requested_channels': requested_count,
            'timed_out_channels': len(pending),
            'non_empty_channels': len(visible_results),
            'validation_status': 'complete',
            'validation': validation,
            'validity': validity,
        }
    SEARCH_CACHE[cache_key] = (now, payload)
    return 200, payload


MASTER_REMOTE_BASE = 'https://xibalami-tggssou.hf.space'
MASTER_WORKERS = [
    ('local', 'self'),
    ('w1', 'https://xibalami-tgsou-w1.hf.space'),
    ('w2', 'https://xibalami-tgsou-w2.hf.space'),
    ('tggssou_alt', 'https://5say-tggssou.hf.space'),
    ('w3_alt', 'https://5say-tgsou-w3.hf.space'),
]

# Historical per-channel load hints.  The cost is roughly “links that need
# validation” observed from previous 108-channel searches.  It is intentionally
# mutable: after every master request we blend in the newly observed channel
# result size, so future splits become less round-robin and more load-balanced.
CHANNEL_LOAD_HINTS = {
    'yoyokuakeduanju': 491,
    'xiangnikanj': 34,
    'kduanju': 31,
    'QuarkFree': 29,
    'XiangxiuNBB': 25,
    'vip115hot': 22,
    'yunpanx': 21,
    'TG654TG': 18,
    'SharePanFilms': 17,
    'rjyxfx': 17,
    'tgsearchers7': 17,
    'kkdj001': 16,
    'kuakedongman': 16,
    'ucquark': 16,
    'zyfb123': 16,
    'Netdisk_Movies': 15,
    'PanjClub': 14,
    'tgbokee': 12,
    'leoziyuan': 11,
    'newproductsourcing': 10,
    'solidsexydoll': 10,
    'tianyifc': 10,
    'yydf_hzl': 10,
    'D_wusun': 9,
    'yunpanxunlei': 8,
    'bdwpzhpd': 7,
    'gokuapan': 7,
    'yunpan139': 7,
    'KaiPanshare': 6,
    'movielover8888_film3': 6,
    'ydypzyfx': 6,
    'yunpanNB': 6,
    'QQZYDAPP': 5,
    'gotopan': 5,
    'Baidu_netdisk': 4,
    'BooksRealm': 4,
    'douerpan': 4,
    'xx123pan': 4,
    'yunpanuc': 4,
    'Q66Share': 3,
    'Quark_Movies': 3,
    'yeqingjie_GJG666': 3,
    'Aliyun_4K_Movies': 2,
    'MCPH03': 2,
    'dianying4k': 2,
    'duan_ju': 2,
    'pxyunpanxunlei': 2,
    'taoxgzy': 2,
    'tyysypzypd': 2,
    'QukanMovie': 1,
    'baicaoZY': 1,
    'godupan': 1,
    'guoman4K': 1,
    'jxwpzy': 1,
    'ucshare': 1,
    'wp123zy': 1,
    'xxzlzn': 1,
    'yp123pan': 1,
    'ysxb48': 1,
    'yunpanquark': 1,
    'zdqxm': 1,
}
CHANNEL_LOAD_HISTORY = dict(CHANNEL_LOAD_HINTS)
CHANNEL_LOAD_LOCK = threading.Lock()
CHANNEL_LOAD_ALPHA = 0.35
DEFAULT_CHANNEL_LOAD = 1.0
# Assign the heaviest first channel to a remote worker; local also coordinates
# the request, so keep it as the last tie-breaker for cold starts.
MASTER_LOAD_TIE_ORDER = {'tggssou': 0, 'tggssou_alt': 0, 'w1': 1, 'w2': 2, 'w3': 3, 'w3_alt': 3, 'local': 4}
# Relative validation capacity observed from cold runs.  Slower Spaces get less
# planned validation weight; faster local/w3 can take more.  This is applied on
# top of per-link type weights.
WORKER_VALIDATION_CAPACITY = {
    'local': 1.0,
    'w3_alt': 1.0,
    'w3': 1.0,
    'w2': 1.0,
    'w1': 1.0,
    'tggssou_alt': 1.0,
    'tggssou': 1.0,
}


def first_value(query, key, default=''):
    return (query.get(key) or [default])[0]


def make_worker_query(query, channels, validate=None):
    q = {k: list(v) for k, v in query.items()}
    q['channelUsername'] = [','.join(channels)]
    q['worker'] = ['1']
    if validate is not None:
        q['validate'] = ['1' if validate else '0']
    return q


def empty_worker_payload():
    return {'results': [], 'validation': validation_summary({}, 0), 'validity': {}, 'non_empty_channels': 0, 'searched_channels': 0, 'requested_channels': 0, 'timed_out_channels': 0}


def call_remote_worker(base, query, channels, validate=None):
    if not channels:
        return 200, empty_worker_payload()
    params = {}
    for k, v in query.items():
        if not v:
            continue
        if k in ('channelUsername', 'channels'):
            continue
        params[k] = v[0]
    params['channelUsername'] = ','.join(channels)
    params['worker'] = '1'
    if validate is not None:
        params['validate'] = '1' if validate else '0'
    url = base + '/api/search?' + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={'User-Agent': 'tgsou-master/1.0'})
    with urllib.request.urlopen(req, timeout=240) as resp:
        raw = resp.read()
        return resp.status, json.loads(raw.decode('utf-8', 'ignore'))


def empty_validate_payload():
    return {'validity': {}, 'validation': validation_summary({}, 0), 'checked_links': 0}


def call_remote_validate_worker(base, links, budget=None):
    links = list(dict.fromkeys(links or []))
    if not links:
        return 200, empty_validate_payload()
    payload = {'links': links}
    if budget:
        payload['budget'] = float(budget)
    data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
    req = urllib.request.Request(base + '/api/validate_links', data=data, method='POST', headers={
        'User-Agent': 'tgsou-master/1.0',
        'Content-Type': 'application/json; charset=utf-8',
        'Accept': 'application/json',
    })
    with urllib.request.urlopen(req, timeout=45 if budget else 240) as resp:
        raw = resp.read()
        return resp.status, json.loads(raw.decode('utf-8', 'ignore'))


def result_map_from_payload(payload):
    out = {}
    for result in payload.get('results', []) or []:
        if result and '$$$' in result:
            ch = result.split('$$$', 1)[0]
            out[ch] = result
    return out


def result_link_count(result):
    if not result or '$$$' not in result:
        return 0
    _, content = result.split('$$$', 1)
    return sum(1 for item in content.split('##') if item.strip())


def channel_load(ch):
    with CHANNEL_LOAD_LOCK:
        return float(CHANNEL_LOAD_HISTORY.get(ch, DEFAULT_CHANNEL_LOAD))


def split_channels_by_historical_load(channels, workers):
    shard_channels = {name: [] for name, _ in workers}
    shard_loads = {name: 0.0 for name, _ in workers}
    # Stable order: heavy known channels first, then original order for ties.
    ordered = sorted(enumerate(channels), key=lambda p: (-channel_load(p[1]), p[0]))
    for _, ch in ordered:
        name, _ = min(workers, key=lambda nb: (shard_loads[nb[0]], MASTER_LOAD_TIE_ORDER.get(nb[0], 99)))
        shard_channels[name].append(ch)
        shard_loads[name] += channel_load(ch)
    return shard_channels, shard_loads


LINK_TYPE_WEIGHTS = {
    'quark': 2.2,
    'uc': 2.2,
    'baidu': 1.5,
    '123pan': 1.4,
    '115': 1.5,
    'xunlei': 0.7,
    '189': 0.7,
    '139': 0.7,
    'other': 1.0,
}


def link_validation_weight(link):
    return float(LINK_TYPE_WEIGHTS.get(cloud_type(link), 1.0))


def split_links_evenly(links, workers):
    shard_links = {name: [] for name, _ in workers}
    shard_weights = {name: 0.0 for name, _ in workers}
    ordered = sorted(enumerate(links or []), key=lambda p: (-link_validation_weight(p[1]), p[0]))
    for _, link in ordered:
        def effective_load(nb):
            name = nb[0]
            cap = float(WORKER_VALIDATION_CAPACITY.get(name, 1.0)) or 1.0
            return (shard_weights[name] / cap, MASTER_LOAD_TIE_ORDER.get(name, 99))
        name, _ = min(workers, key=effective_load)
        shard_links[name].append(link)
        shard_weights[name] += link_validation_weight(link)
    return shard_links, shard_weights


def update_channel_load_history(merged_results):
    observed = {}
    for result in merged_results:
        if result and '$$$' in result:
            ch = result.split('$$$', 1)[0]
            observed[ch] = result_link_count(result)
    if not observed:
        return
    with CHANNEL_LOAD_LOCK:
        for ch, cnt in observed.items():
            old = float(CHANNEL_LOAD_HISTORY.get(ch, DEFAULT_CHANNEL_LOAD))
            CHANNEL_LOAD_HISTORY[ch] = round(old * (1.0 - CHANNEL_LOAD_ALPHA) + float(cnt) * CHANNEL_LOAD_ALPHA, 3)


def merge_validation(payloads):
    validity = {}
    for pld in payloads:
        validity.update(pld.get('validity') or {})
    return validity, validation_summary(validity, len(validity))


def handle_search_master(query):
    keyword = first_value(query, 'keyword').strip()
    raw = first_value(query, 'channelUsername') or first_value(query, 'channels')
    channels = normalize_channels(raw)
    requested_count = len(channels)
    mode_param = first_value(query, 'mode', first_value(query, 'search_mode', 'full')).lower()
    full_requested = mode_param not in ('fast', 'quick')
    priority = [ch for ch in PRIORITY_CHANNELS if ch in channels]
    rest = [ch for ch in channels if ch not in priority]
    ordered = priority + rest
    if full_requested:
        channels = ordered[:FULL_MAX_CHANNELS]
        mode_name = 'full'
    else:
        channels = ordered[:FAST_MAX_CHANNELS]
        mode_name = 'fast'
    if not keyword or not channels:
        return 400, {'error': "Missing 'keyword' or 'channelUsername' parameter"}

    async_validate = first_value(query, 'async_validate', 'false').lower() in ('1', 'true', 'yes', 'on')
    exact_match = first_value(query, 'exact', 'true').lower() not in ('0', 'false', 'no', 'off', 'fuzzy')
    exact_mode = 'exact' if exact_match else 'fuzzy'
    cache_key = (keyword, ','.join(channels), mode_name, 'master_twostage_async_' + exact_mode + '_alt5_smartlocal_compact_xunleiapi_v23' if async_validate else 'master_twostage_sync_' + exact_mode + '_alt5_scheme3_budget25_diskshort_tianyi_xunleiapi_recheck_v34')
    now = time.time()
    cached = SEARCH_CACHE.get(cache_key)
    if cached and now - cached[0] < search_cache_ttl(cached[1]):
        return 200, cached[1]

    # Stage 1: split channel search.  Workers search only, without validating;
    # validation is done in Stage 2 after the master knows the real unique links.
    active_workers = MASTER_WORKERS
    shard_channels, planned_loads = split_channels_by_historical_load(channels, active_workers)
    search_errors = []
    search_payloads = []
    search_infos = []

    def run_search_shard(name, base, part):
        t0 = time.time()
        try:
            if not part:
                st, pld = 200, empty_worker_payload()
            elif base == 'self':
                st, pld = handle_search_local(make_worker_query(query, part, validate=False))
            else:
                st, pld = call_remote_worker(base, query, part, validate=False)
            return name, base, st, pld, time.time() - t0, None
        except Exception as e:
            return name, base, 0, empty_worker_payload(), time.time() - t0, repr(e)

    with ThreadPoolExecutor(max_workers=len(active_workers)) as pool:
        futures = [pool.submit(run_search_shard, name, base, shard_channels.get(name, [])) for name, base in active_workers]
        for fut in as_completed(futures):
            name, base, st, pld, elapsed, err = fut.result()
            info = {'shard': name, 'base': base, 'status': st, 'elapsed_sec': round(elapsed, 3), 'channels': len(shard_channels.get(name, [])), 'planned_load': round(planned_loads.get(name, 0.0), 3), 'non_empty_channels': pld.get('non_empty_channels'), 'validation': pld.get('validation')}
            if err:
                info['error'] = err
                search_errors.append({'phase': 'search', 'shard': name, 'base': base, 'error': err})
            elif st != 200:
                search_errors.append({'phase': 'search', 'shard': name, 'status': st, 'base': base})
            else:
                search_payloads.append(pld)
            search_infos.append(info)

    # If any remote search shard failed, run its channels locally to preserve coverage.
    failed_search = {e.get('shard') for e in search_errors}
    for name, base in active_workers:
        if name in failed_search and base != 'self' and shard_channels.get(name):
            t0 = time.time()
            st, pld = handle_search_local(make_worker_query(query, shard_channels[name], validate=False))
            search_payloads.append(pld)
            search_infos.append({'shard': name + '_fallback_local', 'base': 'self', 'status': st, 'elapsed_sec': round(time.time()-t0, 3), 'channels': len(shard_channels[name]), 'planned_load': round(planned_loads.get(name, 0.0), 3), 'non_empty_channels': pld.get('non_empty_channels'), 'validation': pld.get('validation')})

    maps = [result_map_from_payload(p) for p in search_payloads]
    raw_results = []
    for ch in channels:
        val = ''
        for mp in maps:
            if ch in mp:
                val = mp[ch]
                break
        raw_results.append(val)

    raw_links = extract_links_from_results(raw_results)
    validation_errors = []
    validation_infos = []
    validation_planned_weights = {}

    if async_validate:
        # Async mode keeps the old background validation behavior on the master.
        task_id, task = start_validation_task(raw_results)
        merged_results = compact_results(raw_results)
        validation = {**task.get('validation', {}), 'task_id': task_id, 'poll_url': '/api/validation/' + task_id}
        validity = None
        validation_status = 'processing'
    else:
        # Stage 2: split the actual unique links by estimated validation cost
        # rather than raw count, because Quark/UC/Baidu are much slower than
        # Xunlei/139/189.
        link_shards, validation_planned_weights = split_links_evenly(raw_links, active_workers)
        validation_payloads = []

        def run_validation_shard(name, base, part):
            t0 = time.time()
            try:
                if not part:
                    st, pld = 200, empty_validate_payload()
                elif base == 'self':
                    val, summary = validate_links_budgeted(part, MASTER_SYNC_VALIDATE_BUDGET)
                    pld = {'validity': val, 'validation': summary, 'checked_links': len(part)}
                    st = 200
                else:
                    st, pld = call_remote_validate_worker(base, part, budget=MASTER_SYNC_VALIDATE_BUDGET)
                return name, base, st, pld, time.time() - t0, None
            except Exception as e:
                val = {link: {'status': 'unknown', 'ok': None, 'type': cloud_type(link), 'reason': 'validation_shard_timeout'} for link in (part or [])}
                summary = validation_summary(val, len(part or []))
                summary['budgeted'] = True
                summary['timed_out'] = len(part or [])
                pld = {'validity': val, 'validation': summary, 'checked_links': len(part or []), 'warning': repr(e)}
                return name, base, 200, pld, time.time() - t0, None

        with ThreadPoolExecutor(max_workers=len(active_workers)) as pool:
            futures = [pool.submit(run_validation_shard, name, base, link_shards.get(name, [])) for name, base in active_workers]
            for fut in as_completed(futures):
                name, base, st, pld, elapsed, err = fut.result()
                part_count = len(link_shards.get(name, []))
                info = {'shard': name, 'base': base, 'status': st, 'elapsed_sec': round(elapsed, 3), 'links': part_count, 'planned_weight': round(validation_planned_weights.get(name, 0.0), 3), 'validation': pld.get('validation')}
                if err:
                    info['error'] = err
                    validation_errors.append({'phase': 'validation', 'shard': name, 'base': base, 'links': part_count, 'error': err})
                elif st != 200:
                    validation_errors.append({'phase': 'validation', 'shard': name, 'status': st, 'base': base, 'links': part_count})
                else:
                    validation_payloads.append(pld)
                validation_infos.append(info)

        # If remote validation failed, retry its link shard locally.
        failed_validation = {e.get('shard') for e in validation_errors}
        for name, base in active_workers:
            if name in failed_validation and base != 'self' and link_shards.get(name):
                t0 = time.time()
                val, summary = validate_links_sync(link_shards[name])
                pld = {'validity': val, 'validation': summary, 'checked_links': len(link_shards[name])}
                validation_payloads.append(pld)
                validation_infos.append({'shard': name + '_fallback_local', 'base': 'self', 'status': 200, 'elapsed_sec': round(time.time()-t0, 3), 'links': len(link_shards[name]), 'planned_weight': round(validation_planned_weights.get(name, 0.0), 3), 'validation': summary})

        validity = {}
        for pld in validation_payloads:
            validity.update(pld.get('validity') or {})
        # Remote validation workers may still run older 123pan logic.  Recheck
        # unknown 123pan links on the master so cancelled shares are filtered.
        retry_unknown_123pan_validity(validity)
        retry_xunlei_validity(validity)
        validation = validation_summary(validity, len(raw_links))
        validation['budgeted'] = True
        validation['budget_sec'] = MASTER_SYNC_VALIDATE_BUDGET
        validation['timed_out'] = sum(((pld.get('validation') or {}).get('timed_out') or 0) for pld in validation_payloads)
        merged_results = compact_results(filter_results_by_validity(raw_results, validity))
        validation_status = 'complete'

    update_channel_load_history(raw_results)
    payload = {
        'results': merged_results,
        'cached': False,
        'mode': mode_name,
        'searched_channels': len(channels),
        'requested_channels': requested_count,
        'timed_out_channels': sum((p.get('timed_out_channels') or 0) for p in search_payloads),
        'non_empty_channels': sum(bool(x) for x in merged_results),
        'validation_status': validation_status,
        'validation': validation,
        'master': {
            'enabled': True,
            'strategy': 'two_stage_search_then_scheme3_budget25_diskshort_tianyi_xunleiapi_recheck_validation_%dshards_v12' % len(MASTER_WORKERS),
            'local_base': 'self',
            'workers': [{'name': n, 'base': b} for n, b in MASTER_WORKERS],
            'planned_loads': {name: round(planned_loads.get(name, 0.0), 3) for name, _ in MASTER_WORKERS},
            'validation_planned_weights': {name: round(validation_planned_weights.get(name, 0.0), 3) for name, _ in MASTER_WORKERS},
            'raw_total_links': len(raw_links),
            'shards': search_infos,
            'search_shards': search_infos,
            'validation_shards': validation_infos,
            'errors': search_errors + validation_errors,
        }
    }

    if not async_validate:
        payload['validity'] = validity
    else:
        payload['task_id'] = task_id
    SEARCH_CACHE[cache_key] = (time.time(), payload)
    return 200, payload


def handle_validate_links_request(payload_in):
    raw_links = payload_in.get('links') if isinstance(payload_in, dict) else []
    budget = None
    if isinstance(payload_in, dict):
        try:
            budget = float(payload_in.get('budget')) if payload_in.get('budget') is not None else None
        except Exception:
            budget = None
    if isinstance(raw_links, str):
        parts = re.split(r'[,\n\r\s]+', raw_links)
    elif isinstance(raw_links, list):
        parts = raw_links
    else:
        parts = []
    links = []
    for raw in parts:
        link = clean_url(str(raw or ''))
        if link and is_cloud_url(link) and not is_excluded_cloud_url(link):
            links.append(link)
    links = list(dict.fromkeys(links))[:2000]
    if budget:
        validity, validation = validate_links_budgeted(links, budget)
    else:
        validity, validation = validate_links_sync(links)
    return 200, {'validity': validity, 'validation': validation, 'checked_links': len(links)}


def log_recent_request(method, parsed, query, body_keys=None):
    try:
        item = {
            'ts': int(time.time()),
            'method': method,
            'path': parsed.path,
            'query': {k: v[:3] for k, v in (query or {}).items()},
            'body_keys': body_keys or [],
            'ua': ''
        }
        with REQUEST_LOG_LOCK:
            RECENT_REQUESTS.append(item)
            if len(RECENT_REQUESTS) > REQUEST_LOG_LIMIT:
                del RECENT_REQUESTS[:-REQUEST_LOG_LIMIT]
    except Exception:
        pass


def handle_search(query):
    worker = first_value(query, 'worker').lower() in ('1', 'true', 'yes', 'on')
    split_param = first_value(query, 'split', 'true').lower()
    split = split_param not in ('0', 'false', 'no', 'off')
    aggregate = first_value(query, 'aggregate', 'true').lower() not in ('0', 'false', 'no', 'off')
    if worker or not aggregate or not split:
        return handle_search_local(query)
    return handle_search_master(query)


class Handler(BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'

    def send_bytes(self, status, body, content_type='application/octet-stream', extra_headers=None):
        self.send_response(status)
        self.send_header('Content-Type', content_type)
        self.send_header('Content-Length', str(len(body)))
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', '*')
        if extra_headers:
            for k, v in extra_headers.items():
                if k.lower() not in ('content-length', 'transfer-encoding', 'connection'):
                    self.send_header(k, v)
        self.end_headers()
        if self.command != 'HEAD':
            self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_bytes(204, b'', 'text/plain')

    def do_HEAD(self):
        self.do_GET()

    def do_GET(self):
        parsed = urllib.parse.urlsplit(self.path)
        query = urllib.parse.parse_qs(parsed.query)
        if parsed.path == '/__recent_requests':
            with REQUEST_LOG_LOCK:
                payload = {'ok': True, 'requests': list(RECENT_REQUESTS)}
            body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
            self.send_bytes(200, body, 'application/json; charset=utf-8')
            return
        if parsed.path == '/__clear_recent_requests':
            with REQUEST_LOG_LOCK:
                RECENT_REQUESTS.clear()
            body = json.dumps({'ok': True}, ensure_ascii=False).encode('utf-8')
            self.send_bytes(200, body, 'application/json; charset=utf-8')
            return
        log_recent_request(self.command, parsed, query)
        if parsed.path.startswith('/api/validation/'):
            task_id = parsed.path.rsplit('/', 1)[-1]
            with TASK_LOCK:
                task = VALIDATION_TASKS.get(task_id)
            if not task:
                body = json.dumps({'error': 'task not found'}, ensure_ascii=False).encode('utf-8')
                self.send_bytes(404, body, 'application/json; charset=utf-8')
                return
            payload = task_response(task, include_validity=True)
            body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
            self.send_bytes(200, body, 'application/json; charset=utf-8')
            return
        if parsed.path == '/api/search' or (parsed.path == '/' and 'keyword' in query):
            status, payload = handle_search(query)
            body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
            self.send_bytes(status, body, 'application/json; charset=utf-8')
            return
        if parsed.path.startswith('/api/proxy/'):
            body = json.dumps({'error': 'api proxy disabled'}, ensure_ascii=False).encode('utf-8')
            self.send_bytes(403, body, 'application/json; charset=utf-8')
            return
        if parsed.path == '/healthz':
            self.send_bytes(200, b'ok', 'text/plain')
            return
        self.forward_static(parsed)

    def do_POST(self):
        parsed = urllib.parse.urlsplit(self.path)
        if parsed.path == '/api/validate_links':
            length = int(self.headers.get('Content-Length') or '0')
            raw = self.rfile.read(length) if length else b''
            payload_in = {}
            try:
                payload_in = json.loads(raw.decode('utf-8', 'ignore')) if raw else {}
            except Exception:
                payload_in = {'links': raw.decode('utf-8', 'ignore') if raw else ''}
            status, payload = handle_validate_links_request(payload_in)
            body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
            self.send_bytes(status, body, 'application/json; charset=utf-8')
            return
        if parsed.path == '/api/search' or parsed.path == '/':
            length = int(self.headers.get('Content-Length') or '0')
            raw = self.rfile.read(length) if length else b''
            payload_in = {}
            ctype = (self.headers.get('Content-Type') or '').lower()
            try:
                if 'application/json' in ctype and raw:
                    payload_in = json.loads(raw.decode('utf-8', 'ignore'))
                elif raw:
                    payload_in = {k: v[0] for k, v in urllib.parse.parse_qs(raw.decode('utf-8', 'ignore')).items()}
            except Exception:
                payload_in = {}
            query = urllib.parse.parse_qs(parsed.query)
            for k, v in payload_in.items():
                if v is None:
                    continue
                query[k] = [','.join(str(x) for x in v)] if isinstance(v, list) else [str(v)]
            log_recent_request(self.command, parsed, query, sorted(payload_in.keys()))
            status, payload = handle_search(query)
            body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
            self.send_bytes(status, body, 'application/json; charset=utf-8')
            return
        body = json.dumps({'error': 'not found'}, ensure_ascii=False).encode('utf-8')
        self.send_bytes(404, body, 'application/json; charset=utf-8')

    def forward_static(self, parsed):
        if parsed.path == '/':
            self.send_response(302)
            self.send_header('Location', '/search')
            self.send_header('Content-Length', '0')
            self.end_headers()
            return
        if parsed.path == '/search':
            html = b'<html><head><meta charset="utf-8"><title>tgsou API</title></head><body><h1>tgsou API</h1><p>Use /api/search?keyword=...&channelUsername=...</p><p>Validation is synchronous and included in /api/search response.</p></body></html>'
            self.send_bytes(200, html, 'text/html; charset=utf-8')
            return
        body = b'404 page not found'
        self.send_bytes(404, body, 'text/plain')

    def log_message(self, fmt, *args):
        return


if __name__ == '__main__':
    print(f'front server listening on 0.0.0.0:{PORT}', flush=True)
    ThreadingHTTPServer(('0.0.0.0', PORT), Handler).serve_forever()