211 lines
9.8 KiB
Python
211 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Local controller: serves UI and authenticated browser bridge on loopback only."""
|
|
import argparse
|
|
import hmac
|
|
import importlib.util
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
import secrets
|
|
import sys
|
|
import threading
|
|
from urllib.parse import urlsplit
|
|
import webbrowser
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(ROOT / 'modules/downloader'))
|
|
spec = importlib.util.spec_from_file_location('news_downloader', ROOT / 'modules/downloader/app.py')
|
|
downloader = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(downloader)
|
|
from store import Store
|
|
|
|
|
|
class Controller:
|
|
def __init__(self, data=ROOT / 'data', downloads=ROOT / 'downloads', ledger=ROOT / '发布去重记录.json'):
|
|
self.store = Store(Path(data) / 'queue.sqlite3', ledger)
|
|
self.ui_token, self.bridge_token = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
|
|
self.manager = downloader.Downloader(downloads, on_saved=self.saved)
|
|
self.scan_lock = threading.Lock()
|
|
|
|
def saved(self, article, result, keys):
|
|
directory = Path(result.directory)
|
|
marker = directory / '.article.json'
|
|
pending = directory / '.article.json.tmp'
|
|
pending.write_text(json.dumps({'fingerprints': keys, 'source_url': result.source_url}, ensure_ascii=False), encoding='utf-8')
|
|
pending.replace(marker)
|
|
self.store.enqueue(directory)
|
|
|
|
def scan(self, root):
|
|
if self.manager.thread and self.manager.thread.is_alive():
|
|
raise ValueError('下载进行中，请等待完成后扫描；新下载文章会自动入队')
|
|
root = Path(root).expanduser().resolve(strict=True)
|
|
if not root.is_dir():
|
|
raise ValueError('文章根目录不存在')
|
|
if not self.scan_lock.acquire(blocking=False):
|
|
raise ValueError('正在扫描，请稍候')
|
|
try:
|
|
directories = [root] if list(root.glob('*.md')) else sorted(p for p in root.iterdir() if p.is_dir() and not p.name.startswith('.'))
|
|
result = []
|
|
for directory in directories:
|
|
try:
|
|
status = self.store.enqueue(directory)
|
|
except (ValueError, OSError) as error:
|
|
status = str(error)
|
|
result.append({'directory': directory.name, 'status': status})
|
|
return {'results': result}
|
|
finally:
|
|
self.scan_lock.release()
|
|
|
|
def ui(self, route, body):
|
|
if route == 'state':
|
|
return {**self.store.state(), 'download': self.manager.snapshot()}
|
|
if route == 'pair':
|
|
return {'token': self.bridge_token}
|
|
if route == 'channels':
|
|
key = str(body.get('key', '')).strip()
|
|
if not key or len(key) > 512:
|
|
raise ValueError('请填写 API Key')
|
|
items = downloader.result_list(downloader.api_request(key, 'channel'))
|
|
if not items or not all(isinstance(x, str) and x for x in items):
|
|
raise ValueError('频道响应格式不正确')
|
|
return {'channels': items}
|
|
if route == 'download':
|
|
key, channel, target = str(body.get('key', '')).strip(), body.get('channel'), body.get('target')
|
|
if not key or len(key) > 512 or not isinstance(channel, str) or not channel.strip() or type(target) is not int or not 1 <= target <= 100000:
|
|
raise ValueError('请填写 Key、频道和有效下载数量')
|
|
self.manager.start(key, channel, target)
|
|
return {'ok': True}
|
|
if route == 'download-stop':
|
|
self.manager.stop.set()
|
|
return {'ok': True}
|
|
if route == 'scan':
|
|
return self.scan(body.get('root', '')) if body.get('root') else self.scan(self.manager.root)
|
|
if route == 'configure':
|
|
self.store.configure(body)
|
|
return {'ok': True}
|
|
if route in ('start', 'pause', 'stop'):
|
|
if route == 'stop':
|
|
self.manager.stop.set()
|
|
return self.store.control(route)
|
|
if route == 'reconcile':
|
|
return self.store.reconcile(body)
|
|
if route == 'resolve':
|
|
return self.store.resolve(body)
|
|
if route == 'remove':
|
|
return self.store.remove(body)
|
|
if route == 'history':
|
|
records = body.get('records')
|
|
if not isinstance(records, list):
|
|
raise ValueError('历史记录需包含 records 数组')
|
|
self.store.import_history(records)
|
|
return {'ok': True}
|
|
raise ValueError('未知操作')
|
|
|
|
def bridge(self, route, body):
|
|
methods = {'heartbeat': self.store.heartbeat, 'next': self.store.next, 'package': self.store.package,
|
|
'event': self.store.event, 'accept': self.store.accept, 'switch-begin': self.store.switch_begin,
|
|
'takeover': self.store.takeover, 'current': self.store.current, 'skip': self.store.skip}
|
|
if route in ('start', 'pause', 'stop'):
|
|
if route == 'stop':
|
|
self.manager.stop.set()
|
|
return self.store.control(route)
|
|
if route not in methods:
|
|
raise ValueError('未知浏览器操作')
|
|
return methods[route](body)
|
|
|
|
|
|
def make_handler(controller):
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass # Never echo request bodies, API keys or pairing tokens.
|
|
|
|
def send(self, code, payload, mime='application/json; charset=utf-8'):
|
|
content = payload if isinstance(payload, bytes) else json.dumps(payload, ensure_ascii=False).encode()
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', mime)
|
|
self.send_header('Content-Length', str(len(content)))
|
|
self.send_header('Cache-Control', 'no-store')
|
|
self.send_header('X-Content-Type-Options', 'nosniff')
|
|
self.send_header('Content-Security-Policy', "default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'")
|
|
self.end_headers()
|
|
try:
|
|
self.wfile.write(content)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
pass
|
|
|
|
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 origin_ok(self):
|
|
origin = self.headers.get('Origin')
|
|
return not origin or origin in (f'http://127.0.0.1:{self.server.server_port}', f'http://localhost:{self.server.server_port}')
|
|
|
|
def do_GET(self):
|
|
if not self.local() or not self.origin_ok():
|
|
return self.send(403, {'error': '仅允许本机控制台访问'})
|
|
route = urlsplit(self.path).path
|
|
if route == '/api/session':
|
|
return self.send(200, {'token': controller.ui_token})
|
|
files = {'/': ('static/index.html', 'text/html; charset=utf-8'),
|
|
'/app.js': ('static/app.js', 'text/javascript; charset=utf-8'),
|
|
'/style.css': ('static/style.css', 'text/css; charset=utf-8'),
|
|
'/browser/auto-news.user.js': ('browser/auto-news.user.js', 'text/javascript; charset=utf-8'),
|
|
'/browser/local-bridge.user.js': ('browser/local-bridge.user.js', 'text/javascript; charset=utf-8')}
|
|
if route not in files:
|
|
return self.send(404, {'error': '不存在'})
|
|
name, mime = files[route]
|
|
file = ROOT / name
|
|
if not file.exists():
|
|
return self.send(404, {'error': '请先运行 build_browser.py'})
|
|
return self.send(200, file.read_bytes(), mime)
|
|
|
|
def do_POST(self):
|
|
if not self.local():
|
|
return self.send(403, {'error': 'Host 不匹配'})
|
|
route = urlsplit(self.path).path
|
|
bridge = route.startswith('/bridge/')
|
|
expected = controller.bridge_token if bridge else controller.ui_token
|
|
provided = self.headers.get('X-Session-Token', '')
|
|
if not hmac.compare_digest(expected, provided) or (not bridge and not self.origin_ok()):
|
|
return self.send(403, {'error': '连接凭据无效，请重新配对'})
|
|
if not bridge and not route.startswith('/api/'):
|
|
return self.send(404, {'error': '不存在'})
|
|
try:
|
|
length = int(self.headers.get('Content-Length', 0))
|
|
if not 0 < length <= 2 * 1024 * 1024:
|
|
raise ValueError('请求大小无效')
|
|
body = json.loads(self.rfile.read(length))
|
|
if not isinstance(body, dict):
|
|
raise ValueError('请求必须是对象')
|
|
result = (controller.bridge if bridge else controller.ui)(route.rsplit('/', 1)[1], body)
|
|
self.send(200, result)
|
|
except (ValueError, KeyError, OSError) as error:
|
|
self.send(400, {'error': str(error)[:1000]})
|
|
except Exception:
|
|
self.send(500, {'error': '内部错误；请停止当前任务并检查本地服务'})
|
|
return Handler
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--port', type=int, default=8770)
|
|
parser.add_argument('--no-browser', action='store_true')
|
|
args = parser.parse_args()
|
|
controller = Controller()
|
|
server = ThreadingHTTPServer(('127.0.0.1', args.port), make_handler(controller))
|
|
url = f'http://127.0.0.1:{args.port}/'
|
|
print('新闻工作台：' + url, flush=True)
|
|
if not args.no_browser:
|
|
webbrowser.open(url)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
controller.manager.stop.set()
|
|
controller.store.control('stop')
|
|
finally:
|
|
server.server_close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|