#!/usr/bin/env python3
"""仅在本机运行的探数新闻下载器。"""
from collections import deque
from datetime import datetime
from hashlib import sha256
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
import argparse
import json
import secrets
import sqlite3
import threading
import time
import unicodedata
import webbrowser

import requests
from bs4 import BeautifulSoup
from article_writer import save_article

ROOT = Path(__file__).resolve().parent
API = 'https://api.tanshuapi.com/api/toutiao/v1/'


def api_request(key, endpoint, params=None):
    if endpoint not in ('channel', 'index'):
        raise ValueError('未知接口')
    try:
        # 不自动重试计费接口，避免一次操作产生不可见的重复调用。
        response = requests.get(API + endpoint, params={'key': key, **(params or {})}, timeout=(10, 40))
        response.raise_for_status()
        payload = response.json()
    except requests.RequestException:
        raise ValueError('接口网络请求失败或超时，请在调试终端检查后重试') from None
    except ValueError:
        raise ValueError('接口未返回有效 JSON') from None
    if not isinstance(payload, dict):
        raise ValueError('接口响应格式错误')
    return payload


def result_list(payload):
    if payload.get('code') != 1:
        raise ValueError(f"探数接口错误 {payload.get('code')}：{payload.get('msg', '未知错误')}")
    data = payload.get('data')
    if not isinstance(data, dict) or not isinstance(data.get('list'), list):
        raise ValueError('接口响应缺少 data.list')
    return data['list']


def fingerprints(article):
    values = []
    for field in ('url', 'weburl'):
        value = str(article.get(field) or '').strip()
        if value:
            u = urlsplit(value)
            query = [(k, v) for k, v in parse_qsl(u.query) if not k.lower().startswith('utm_') and k.lower() not in ('spm', 'from', 'pos', 'vt')]
            values.append('url:' + urlunsplit(('', u.netloc.lower(), u.path, urlencode(sorted(query)), '')))
    body = BeautifulSoup(str(article.get('content') or ''), 'html.parser').get_text(' ', strip=True)
    body = ''.join(unicodedata.normalize('NFKC', body).split())
    if body:
        values.append('body:' + body)
    return [sha256(value.encode()).hexdigest() for value in values]


class Downloader:
    def __init__(self, root=ROOT / 'downloads', requester=api_request, writer=save_article):
        self.root = Path(root).resolve()
        self.root.mkdir(parents=True, exist_ok=True)
        self.requester, self.writer = requester, writer
        self.lock = threading.RLock()
        self.stop = threading.Event()
        self.thread = None
        self.logs = deque(maxlen=300)
        self.seq = 0
        self.state = dict(status='idle', saved=0, target=0, skipped=0, failed=0, scanned=0, requests=0, channel='', message='等待开始')
        with self.db() as db:
            db.execute('CREATE TABLE IF NOT EXISTS seen (fingerprint TEXT PRIMARY KEY, directory TEXT NOT NULL)')

    def db(self):
        return sqlite3.connect(self.root / '.history.sqlite3')

    def log(self, message):
        with self.lock:
            self.seq += 1
            self.logs.append(dict(id=self.seq, time=datetime.now().strftime('%H:%M:%S'), message=message))

    def update(self, **values):
        with self.lock:
            self.state.update(values)

    def snapshot(self):
        with self.lock:
            return {**self.state, 'logs': list(self.logs), 'output': str(self.root)}

    def start(self, key, channel, target):
        with self.lock:
            if self.thread and self.thread.is_alive():
                raise ValueError('已有任务正在运行')
            self.stop.clear()
            self.logs.clear()
            self.state = dict(status='running', saved=0, target=target, skipped=0, failed=0, scanned=0, requests=0, channel=channel, message='正在下载')
            self.thread = threading.Thread(target=self.run, args=(key, channel, target), daemon=True)
            self.thread.start()

    def run(self, key, channel, target):
        saved = skipped = failed = scanned = calls = 0
        seen_in_run = set()
        self.log(f'开始下载频道「{channel}」，目标 {target} 条；历史重复新闻将跳过')
        try:
            with self.db() as db:
                offset = 0
                while offset <= 400 and saved < target and not self.stop.is_set():
                    count = min(40, target - saved)
                    calls += 1
                    self.update(requests=calls)
                    self.log(f'请求第 {calls} 页：start={offset}，num={count}')
                    items = result_list(self.requester(key, 'index', {'type': channel, 'num': count, 'start': offset}))
                    if not items:
                        self.log('接口返回空列表，已停止')
                        break
                    for article in items:
                        if saved >= target or self.stop.is_set():
                            break
                        scanned += 1
                        try:
                            if not isinstance(article, dict):
                                raise ValueError('新闻条目格式错误')
                            keys = fingerprints(article)
                            existing = any(k in seen_in_run for k in keys)
                            for k in keys:
                                row = db.execute('SELECT directory FROM seen WHERE fingerprint=?', (k,)).fetchone()
                                if row and Path(row[0]).is_dir():
                                    existing = True
                            seen_in_run.update(keys)
                            if existing:
                                skipped += 1
                                self.log('跳过重复新闻：' + str(article.get('title', '无标题'))[:160])
                                continue
                            result = self.writer(article, self.root)
                            try:
                                db.executemany('INSERT OR REPLACE INTO seen VALUES (?, ?)', [(k, result.directory) for k in keys])
                                db.commit()
                            except Exception:
                                # 保存目录尚未登记成功，回滚本次产物，避免下次重复保存。
                                import shutil
                                shutil.rmtree(result.directory, ignore_errors=True)
                                raise
                            saved += 1
                            self.log(f'下载新闻 {saved}/{target} 条：{result.title}')
                        except Exception as exc:
                            failed += 1
                            self.log(f'保存失败：{str(exc).replace(key, "[已隐藏]")[:400]}')
                        finally:
                            self.update(saved=saved, skipped=skipped, failed=failed, scanned=scanned)
                    # 避免越过 start=400；已消费的条目决定下一起始位置。
                    offset += len(items)
                    if offset > 400:
                        self.log('已到达接口允许的分页范围')
                    if saved < target and offset <= 400:
                        self.stop.wait(0.3)
            status = 'completed' if saved >= target else 'cancelled' if self.stop.is_set() else 'partial'
            message = f'已保存 {saved}/{target} 条，跳过重复 {skipped} 条，失败 {failed} 条'
            if status == 'partial':
                message += '；当前可获取新闻不足，未达到目标数量'
            if status == 'cancelled':
                message += '；任务已停止'
            self.update(status=status, message=message)
            self.log(message)
        except Exception as exc:
            message = str(exc).replace(key, '[已隐藏]')[:500]
            self.update(status='error', message=message)
            self.log('任务中断：' + message)


def make_handler(manager, token):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def send(self, status, payload, mime='application/json; charset=utf-8'):
            data = json.dumps(payload, ensure_ascii=False).encode() if isinstance(payload, dict) else payload
            self.send_response(status)
            self.send_header('Content-Type', mime)
            self.send_header('Content-Length', str(len(data)))
            self.send_header('Cache-Control', 'no-store')
            self.send_header('X-Content-Type-Options', 'nosniff')
            self.end_headers()
            self.wfile.write(data)

        def local(self):
            return self.headers.get('Host') in (f'127.0.0.1:{self.server.server_port}', f'localhost:{self.server.server_port}')

        def do_GET(self):
            if not self.local():
                return self.send(403, {'error': '仅允许本机访问'})
            path = urlsplit(self.path).path
            if path == '/api/state':
                return self.send(200, manager.snapshot())
            if path == '/api/session':
                return self.send(200, {'token': token})
            files = {'/': ('index.html', 'text/html; charset=utf-8'), '/app.js': ('app.js', 'text/javascript; charset=utf-8'), '/style.css': ('style.css', 'text/css; charset=utf-8')}
            if path not in files:
                return self.send(404, {'error': '未找到页面'})
            name, mime = files[path]
            self.send(200, (ROOT / 'static' / name).read_bytes(), mime)

        def do_POST(self):
            if not self.local() or self.headers.get('X-Session-Token') != token:
                return self.send(403, {'error': '页面会话失效，请刷新页面'})
            key = ''
            try:
                length = int(self.headers.get('Content-Length', 0))
                if not 0 < length <= 16384:
                    raise ValueError('请求大小无效')
                body = json.loads(self.rfile.read(length))
                if not isinstance(body, dict):
                    raise ValueError('请求必须是 JSON 对象')
                if self.path == '/api/stop':
                    manager.stop.set()
                    return self.send(200, {'ok': True})
                key = str(body.get('key') or '').strip()
                if not key or len(key) > 512:
                    raise ValueError('请填写有效 API Key')
                if self.path == '/api/channels':
                    channels = result_list(manager.requester(key, 'channel'))
                    if not channels or not all(isinstance(c, str) and c for c in channels):
                        raise ValueError('频道列表为空或格式错误')
                    return self.send(200, {'channels': channels})
                if self.path == '/api/start':
                    target = body.get('target')
                    channel = body.get('channel')
                    if type(target) is not int or not 1 <= target <= 100000:
                        raise ValueError('下载数量须为 1 至 100000 的整数')
                    if not isinstance(channel, str) or not channel.strip() or len(channel) > 100:
                        raise ValueError('请选择新闻频道')
                    manager.start(key, channel, target)
                    return self.send(200, {'ok': True})
                if self.path == '/api/debug':
                    endpoint = body.get('endpoint')
                    params = {}
                    if endpoint == 'index':
                        for field, low, high in [('num', 1, 40), ('start', 0, 400)]:
                            value = body.get(field)
                            if type(value) is not int or not low <= value <= high:
                                raise ValueError(f'{field} 必须在 {low}–{high} 之间')
                            params[field] = value
                        params['type'] = str(body.get('channel') or '')
                    before = time.monotonic()
                    payload = manager.requester(key, endpoint, params)
                    # 服务端响应同样隐藏 key，且只作为文本展示，不渲染上游 HTML。
                    clean = json.loads(json.dumps(payload, ensure_ascii=False).replace(key, '[已隐藏]'))
                    return self.send(200, {'elapsed': round(time.monotonic() - before, 2), 'request': API + str(endpoint) + '?' + urlencode({'key': '[已隐藏]', **params}), 'response': clean})
                self.send(404, {'error': '未知操作'})
            except Exception as exc:
                message = str(exc)
                if key:
                    message = message.replace(key, '[已隐藏]')
                self.send(400, {'error': message[:500]})
    return Handler


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--port', type=int, default=8765)
    parser.add_argument('--no-browser', action='store_true')
    args = parser.parse_args()
    manager = Downloader()
    server = ThreadingHTTPServer(('127.0.0.1', args.port), make_handler(manager, secrets.token_urlsafe(32)))
    url = f'http://127.0.0.1:{server.server_port}'
    print(f'新闻下载器：{url}\n保存目录：{manager.root}\n按 Ctrl+C 退出', flush=True)
    if not args.no_browser:
        webbrowser.open(url)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        manager.stop.set()
        if manager.thread:
            manager.thread.join(timeout=45)
    finally:
        server.server_close()


if __name__ == '__main__':
    main()