better-staridc-MNBT
1#!/usr/bin/env python3
2# coding: utf-8
3
4import hashlib
5import hmac
6import json
7import logging
8import logging.handlers
9import os
10import re
11import secrets
12import sqlite3
13import sys
14import threading
15import time
16import urllib.error
17import urllib.request
18from concurrent.futures import ThreadPoolExecutor, as_completed
19from datetime import datetime
20from urllib.parse import urlparse
21from stats_collector import init_stats_db, collect_site_stats
22
23
24BASE_DIR = os.path.dirname(os.path.abspath(__file__))
25CONFIG_PATH = os.path.join(BASE_DIR, "config.json")
26INDEX_DB_PATH = os.path.join(BASE_DIR, "file_index.db")
27WORKER_LOG_PATH = os.path.join(BASE_DIR, "worker.log")
28
29_logger = None
30_log_lock = threading.Lock()
31
32SCAN_THREAD_COUNT = min(8, max(2, (os.cpu_count() or 4)))
33SCAN_BATCH_SIZE = 50
34SCAN_PROGRESS_INTERVAL = 5
35MAX_SCAN_MEMORY_MB = 512
36
37
38def get_logger():
39 global _logger
40 if _logger is not None:
41 return _logger
42 logger = logging.getLogger("mnbt_worker")
43 logger.setLevel(logging.DEBUG)
44 logger.propagate = True
45 root_logger = logging.getLogger()
46 if not root_logger.handlers:
47 try:
48 handler = logging.handlers.RotatingFileHandler(
49 WORKER_LOG_PATH,
50 maxBytes=10 * 1024 * 1024,
51 backupCount=5,
52 encoding="utf-8",
53 )
54 except OSError:
55 handler = logging.StreamHandler(sys.stdout)
56 config = {}
57 try:
58 if os.path.exists(CONFIG_PATH):
59 with open(CONFIG_PATH, "r", encoding="utf-8") as handle:
60 config = json.load(handle)
61 except Exception:
62 pass
63 level_name = (config.get("log_level") or "INFO").upper()
64 level_map = {
65 "DEBUG": logging.DEBUG,
66 "INFO": logging.INFO,
67 "WARNING": logging.WARNING,
68 "ERROR": logging.ERROR,
69 }
70 handler.setLevel(level_map.get(level_name, logging.INFO))
71 fmt = logging.Formatter(
72 "%(asctime)s [%(levelname)s] %(message)s",
73 datefmt="%Y-%m-%d %H:%M:%S",
74 )
75 handler.setFormatter(fmt)
76 root_logger.addHandler(handler)
77 root_logger.setLevel(logging.DEBUG)
78 _logger = logger
79 return logger
80
81
82class _StdoutToLogger:
83 def __init__(self, level=logging.INFO):
84 self.level = level
85 self._buf = ""
86
87 def write(self, msg):
88 if not isinstance(msg, str):
89 msg = str(msg)
90 self._buf += msg
91 while "\n" in self._buf:
92 line, self._buf = self._buf.split("\n", 1)
93 if line.strip():
94 logging.getLogger("stdout").log(self.level, line)
95
96 def flush(self):
97 if self._buf.strip():
98 logging.getLogger("stdout").log(self.level, self._buf)
99 self._buf = ""
100
101
102def log_debug(msg, *args):
103 get_logger().debug(msg, *args)
104
105
106def log_info(msg, *args, **kwargs):
107 get_logger().info(msg, *args, **kwargs)
108
109
110def log_warn(msg, *args, **kwargs):
111 get_logger().warning(msg, *args, **kwargs)
112
113
114def log_error(msg, *args, **kwargs):
115 get_logger().error(msg, *args, **kwargs)
116
117
118def setup_worker_log():
119 get_logger()
120 sys.stdout = _StdoutToLogger(logging.INFO)
121 sys.stderr = _StdoutToLogger(logging.ERROR)
122 log_info("worker 日志系统已初始化完成，日志文件：%s", WORKER_LOG_PATH)
123
124# 默认文本扩展名
125TEXT_EXTENSIONS = {
126 ".php", ".html", ".htm", ".js", ".css", ".txt", ".json", ".md",
127 ".xml", ".vue", ".tpl", ".ini", ".conf", ".yml", ".yaml"
128}
129
130# 默认跳过目录
131DEFAULT_SKIP_DIRS = {"cache", "runtime", "logs", ".git", "node_modules", "vendor", "__pycache__"}
132
133# 默认跳过后缀
134DEFAULT_SKIP_EXTS = {".jpg", ".png", ".gif", ".webp", ".mp4", ".zip", ".rar", ".7z", ".pdf", ".woff", ".ttf", ".mp3", ".avi", ".mov"}
135
136
137def now_text():
138 return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
139
140
141def load_config():
142 with open(CONFIG_PATH, "r", encoding="utf-8") as handle:
143 return json.load(handle)
144
145
146def body_hash(body):
147 return hashlib.sha256(body).hexdigest()
148
149
150def signing_key(platform_secret, node_secret):
151 return hmac.new(
152 platform_secret.encode("utf-8"),
153 node_secret.encode("utf-8"),
154 hashlib.sha256,
155 ).hexdigest()
156
157
158def signature(method, path, body, platform_secret, node_secret, timestamp, nonce):
159 canonical = "\n".join([
160 method.upper(),
161 path,
162 body_hash(body),
163 str(timestamp),
164 nonce,
165 ])
166 return hmac.new(
167 signing_key(platform_secret, node_secret).encode("utf-8"),
168 canonical.encode("utf-8"),
169 hashlib.sha256,
170 ).hexdigest()
171
172
173def signed_post(config, action, payload):
174 url = config.get("mnbt_url", "").rstrip("/") + "/api/node.php?act=" + action
175 log_debug("signed_post 开始请求 %s -> %s", action, url)
176 parsed = urlparse(url)
177 path = parsed.path + ("?" + parsed.query if parsed.query else "")
178 body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
179 timestamp = int(time.time())
180 nonce = secrets.token_hex(12)
181 headers = {
182 "Content-Type": "application/json",
183 "X-MNBT-Node": config.get("node_id", ""),
184 "X-MNBT-Time": str(timestamp),
185 "X-MNBT-Nonce": nonce,
186 "X-MNBT-Sign": signature(
187 "POST",
188 path,
189 body,
190 config.get("platform_secret", ""),
191 config.get("node_secret", ""),
192 timestamp,
193 nonce,
194 ),
195 }
196 request = urllib.request.Request(url, data=body, headers=headers, method="POST")
197 try:
198 with urllib.request.urlopen(request, timeout=5) as response:
199 data = response.read().decode("utf-8")
200 log_debug("signed_post %s 成功", action)
201 return json.loads(data)
202 except Exception as e:
203 log_error("signed_post %s 失败: %s", action, e)
204 raise
205
206
207def heartbeat(config):
208 return signed_post(config, "heartbeat", {
209 "node_name": config.get("node_name", ""),
210 "version": config.get("version", "0.1.0"),
211 "capabilities": config.get("capabilities", []),
212 })
213
214
215def get_forbidden_config(config):
216 """从 MNBT 获取违禁词扫描配置"""
217 try:
218 response = signed_post(config, "get_config", {})
219 if not response.get("success"):
220 return None
221 return response.get("data", {}).get("forbidden_scan", {})
222 except Exception as e:
223 print(now_text(), "获取违禁词配置失败：", e, file=sys.stderr)
224 return None
225
226
227def pull_task(config):
228 response = signed_post(config, "pull_task", {})
229 if not response.get("success"):
230 return None
231 return (response.get("data") or {}).get("task")
232
233
234def report_result(config, task_id, status, result=None, error=""):
235 return signed_post(config, "report_result", {
236 "task_id": task_id,
237 "status": status,
238 "result": result or {},
239 "error": error,
240 })
241
242
243# ==================== 文件索引数据库 ====================
244
245def init_index_db():
246 """初始化文件索引 SQLite 数据库"""
247 conn = sqlite3.connect(INDEX_DB_PATH)
248 conn.execute("PRAGMA journal_mode=WAL")
249 cursor = conn.cursor()
250 cursor.execute("""
251 CREATE TABLE IF NOT EXISTS file_index (
252 path TEXT PRIMARY KEY NOT NULL,
253 size INTEGER NOT NULL,
254 mtime REAL NOT NULL,
255 fingerprint TEXT NOT NULL,
256 last_scanned REAL NOT NULL,
257 scan_mode TEXT NOT NULL
258 )
259 """)
260 cursor.execute("CREATE INDEX IF NOT EXISTS idx_fingerprint ON file_index(fingerprint)")
261 cursor.execute("CREATE INDEX IF NOT EXISTS idx_mtime ON file_index(mtime)")
262 conn.commit()
263 conn.close()
264
265
266def get_index_conn():
267 """获取索引数据库连接"""
268 conn = sqlite3.connect(INDEX_DB_PATH)
269 conn.execute("PRAGMA journal_mode=WAL")
270 return conn
271
272
273def compute_file_fingerprint(path):
274 """计算文件指纹: path + size + mtime + sha256"""
275 try:
276 stat_info = os.stat(path)
277 size = stat_info.st_size
278 mtime = stat_info.st_mtime
279
280 # 先快速检查 size + mtime，避免频繁计算 sha256
281 fast_hash = f"{path}|{size}|{mtime}"
282
283 # 对于小文件才计算 sha256（避免大文件拖慢扫描）
284 if size < 1024 * 1024: # 1MB 以下才算 sha256
285 sha256 = hashlib.sha256()
286 with open(path, "rb") as f:
287 for chunk in iter(lambda: f.read(8192), b""):
288 sha256.update(chunk)
289 return f"{fast_hash}|{sha256.hexdigest()}"
290 else:
291 return f"{fast_hash}|{size}"
292 except OSError:
293 return None
294
295
296def get_file_index(path):
297 """获取文件的索引记录"""
298 conn = get_index_conn()
299 cursor = conn.cursor()
300 cursor.execute("SELECT size, mtime, fingerprint, last_scanned, scan_mode FROM file_index WHERE path = ?", (path,))
301 row = cursor.fetchone()
302 conn.close()
303 if row:
304 return {
305 "size": row[0],
306 "mtime": row[1],
307 "fingerprint": row[2],
308 "last_scanned": row[3],
309 "scan_mode": row[4],
310 }
311 return None
312
313
314def update_file_index(path, scan_mode="incremental"):
315 """更新文件索引"""
316 fingerprint = compute_file_fingerprint(path)
317 if not fingerprint:
318 return False
319
320 stat_info = os.stat(path)
321 now = time.time()
322
323 conn = get_index_conn()
324 cursor = conn.cursor()
325 cursor.execute("""
326 INSERT OR REPLACE INTO file_index (path, size, mtime, fingerprint, last_scanned, scan_mode)
327 VALUES (?, ?, ?, ?, ?, ?)
328 """, (path, stat_info.st_size, stat_info.st_mtime, fingerprint, now, scan_mode))
329 conn.commit()
330 conn.close()
331 return True
332
333
334def delete_file_index(path):
335 """删除文件索引"""
336 conn = get_index_conn()
337 cursor = conn.cursor()
338 cursor.execute("DELETE FROM file_index WHERE path = ?", (path,))
339 conn.commit()
340 conn.close()
341
342
343def clean_orphaned_index(root):
344 """清理索引中不存在的文件记录"""
345 conn = get_index_conn()
346 cursor = conn.cursor()
347 cursor.execute("SELECT path FROM file_index")
348 rows = cursor.fetchall()
349 deleted = 0
350 for (path,) in rows:
351 if not os.path.exists(path):
352 cursor.execute("DELETE FROM file_index WHERE path = ?", (path,))
353 deleted += 1
354 conn.commit()
355 conn.close()
356 return deleted
357
358
359# ==================== 扫描逻辑 ====================
360
361def safe_join(root, relative):
362 root = os.path.abspath(root)
363 target = os.path.abspath(os.path.join(root, relative.lstrip("/")))
364 root_norm = os.path.abspath(root).rstrip(os.sep) or os.sep
365 if target != root_norm and not target.startswith(root_norm + os.sep):
366 raise ValueError("文件路径超出扫描根目录范围")
367 return target
368
369
370def parse_skip_dirs(skip_dirs_str):
371 """解析跳过目录字符串"""
372 if not skip_dirs_str:
373 return DEFAULT_SKIP_DIRS.copy()
374 dirs = [d.strip() for d in skip_dirs_str.split(",") if d.strip()]
375 return set(dirs) or DEFAULT_SKIP_DIRS.copy()
376
377
378def parse_skip_exts(skip_exts_str):
379 """解析跳过后缀字符串"""
380 if not skip_exts_str:
381 return DEFAULT_SKIP_EXTS.copy()
382 exts = [e.strip().lower() for e in skip_exts_str.split(",") if e.strip()]
383 exts = [e if e.startswith(".") else "." + e for e in exts]
384 return set(exts) or DEFAULT_SKIP_EXTS.copy()
385
386
387def parse_keywords(content):
388 """解析违禁词内容，每行一个"""
389 keywords = []
390 for line in content.split("\n"):
391 line = line.strip()
392 if line and not line.startswith("#"):
393 keywords.append(line)
394 return keywords
395
396
397def iter_scan_files(root, max_file_size, skip_dirs=None, skip_exts=None):
398 """遍历需要扫描的文件"""
399 if skip_dirs is None:
400 skip_dirs = DEFAULT_SKIP_DIRS.copy()
401 if skip_exts is None:
402 skip_exts = DEFAULT_SKIP_EXTS.copy()
403
404 for current_root, dirs, files in os.walk(root):
405 # 过滤跳过的目录
406 dirs[:] = [item for item in dirs if item not in skip_dirs and not item.startswith(".")]
407
408 for filename in files:
409 ext = os.path.splitext(filename)[1].lower()
410 if ext in skip_exts:
411 continue
412 path = os.path.join(current_root, filename)
413 try:
414 if os.path.getsize(path) > max_file_size:
415 continue
416 yield path
417 except OSError:
418 continue
419
420
421def should_scan_file(path, scan_changed_only=True, scan_mode="incremental"):
422 """判断文件是否需要扫描"""
423 if not scan_changed_only:
424 return True
425
426 # 检查索引
427 index = get_file_index(path)
428 if not index:
429 return True # 新文件，需要扫描
430
431 # 检查文件是否变化
432 current_fingerprint = compute_file_fingerprint(path)
433 if not current_fingerprint:
434 return False
435
436 if current_fingerprint != index["fingerprint"]:
437 return True # 文件已修改，需要扫描
438
439 return False # 文件未变化，跳过
440
441
442def excerpt_for_line(line, keyword, width=80):
443 index = line.find(keyword)
444 if index < 0:
445 return line[:width]
446 start = max(0, index - width // 2)
447 end = min(len(line), index + len(keyword) + width // 2)
448 return line[start:end].strip()
449
450
451def scan_single_file(path, root, keywords, max_matches, scan_mode, _match_lock, _matches, stop_event):
452 """单文件扫描函数（线程池调用），返回 (path, file_matches, error)"""
453 if stop_event.is_set():
454 return path, [], None
455 try:
456 file_matches = []
457 with open(path, "r", encoding="utf-8", errors="ignore") as handle:
458 for line_no, line in enumerate(handle, 1):
459 if stop_event.is_set():
460 break
461 for keyword in keywords:
462 if keyword in line:
463 relative_path = os.path.relpath(path, root)
464 file_matches.append({
465 "site": relative_path,
466 "type": "file",
467 "path": relative_path,
468 "line": line_no,
469 "keyword": keyword,
470 "excerpt": excerpt_for_line(line, keyword),
471 })
472 break
473 try:
474 update_file_index(path, scan_mode)
475 except Exception:
476 pass
477 return path, file_matches, None
478 except Exception as e:
479 return path, [], str(e)
480
481
482def collect_scan_files(root, max_file_size, skip_dirs, skip_exts,
483 scan_changed_only, scan_mode):
484 """收集所有需要扫描的文件路径列表"""
485 all_files = []
486 skipped_index = 0
487 for path in iter_scan_files(root, max_file_size, skip_dirs, skip_exts):
488 if not should_scan_file(path, scan_changed_only, scan_mode):
489 skipped_index += 1
490 continue
491 all_files.append(path)
492 log_info("扫描文件收集完成：待扫描 %d 个，索引跳过 %d 个", len(all_files), skipped_index)
493 return all_files
494
495
496def forbidden_scan_incremental(root, keywords, max_file_size, max_matches,
497 skip_dirs=None, skip_exts=None,
498 scan_changed_only=True, scan_mode="incremental",
499 thread_count=None):
500 """多线程增量扫描违禁词"""
501 root = os.path.abspath(root)
502 if not keywords:
503 raise ValueError("违禁词列表为空")
504 if not os.path.isdir(root):
505 raise ValueError("扫描目录不存在")
506
507 if thread_count is None:
508 thread_count = SCAN_THREAD_COUNT
509
510 log_info("开始%s扫描：root=%s, 线程数=%d, 关键词=%d个, max_matches=%d",
511 "全量" if scan_mode == "full" else "增量",
512 root, thread_count, len(keywords), max_matches)
513
514 start_time = time.time()
515 all_files = collect_scan_files(root, max_file_size, skip_dirs, skip_exts,
516 scan_changed_only, scan_mode)
517 total_files = len(all_files)
518 if total_files == 0:
519 log_info("没有需要扫描的文件，直接结束")
520 return {
521 "site": os.path.basename(root),
522 "summary": {
523 "scanned_files": 0,
524 "scanned_rows": 0,
525 "matches": 0,
526 "finished_at": now_text(),
527 "scan_mode": scan_mode,
528 "duration_sec": 0,
529 "thread_count": thread_count,
530 "total_files": 0,
531 },
532 "matches": [],
533 }
534
535 matches = []
536 match_lock = threading.Lock()
537 stop_event = threading.Event()
538 scanned_count = 0
539 error_count = 0
540 last_progress_time = 0
541
542 def on_file_done(_path, _file_matches):
543 nonlocal scanned_count, error_count, last_progress_time
544 with match_lock:
545 scanned_count += 1
546 if _file_matches:
547 if len(matches) < max_matches:
548 remaining = max_matches - len(matches)
549 matches.extend(_file_matches[:remaining])
550 if len(matches) >= max_matches:
551 stop_event.set()
552 log_info("命中数已达上限 %d，停止扫描", max_matches)
553 now = time.time()
554 if now - last_progress_time >= SCAN_PROGRESS_INTERVAL:
555 last_progress_time = now
556 pct = (scanned_count / total_files) * 100 if total_files > 0 else 0
557 log_info("扫描进度：%d/%d (%.1f%%), 当前命中 %d 条",
558 scanned_count, total_files, pct, len(matches))
559
560 thread_count = min(thread_count, max(1, total_files))
561 log_info("启动 %d 个扫描线程，共 %d 个文件待扫描", thread_count, total_files)
562
563 with ThreadPoolExecutor(max_workers=thread_count, thread_name_prefix="scan_worker") as executor:
564 future_map = {executor.submit(
565 scan_single_file, path, root, keywords, max_matches,
566 scan_mode, match_lock, matches, stop_event
567 ): path for path in all_files}
568
569 for future in as_completed(future_map):
570 path = future_map[future]
571 try:
572 fpath, file_matches, err = future.result()
573 if err:
574 error_count += 1
575 log_warn("文件扫描出错 %s: %s", path, err)
576 on_file_done(fpath, file_matches)
577 except Exception as e:
578 error_count += 1
579 log_error("扫描任务异常 %s: %s", path, e)
580
581 if stop_event.is_set():
582 for f in future_map:
583 f.cancel()
584 break
585
586 duration = time.time() - start_time
587 speed = scanned_count / duration if duration > 0 else 0
588 log_info("扫描完成：扫描文件 %d 个，出错 %d 个，命中 %d 条，耗时 %.1f 秒 (%.1f 文件/秒)",
589 scanned_count, error_count, len(matches), duration, speed)
590
591 return {
592 "site": os.path.basename(root),
593 "summary": {
594 "scanned_files": scanned_count,
595 "scanned_rows": 0,
596 "matches": len(matches),
597 "finished_at": now_text(),
598 "scan_mode": scan_mode,
599 "duration_sec": round(duration, 2),
600 "thread_count": thread_count,
601 "total_files": total_files,
602 "error_count": error_count,
603 "speed_files_per_sec": round(speed, 2),
604 },
605 "matches": matches,
606 }
607
608
609# ==================== 任务执行 ====================
610
611def execute_task(task, config):
612 """执行任务"""
613 action = task.get("action")
614 payload = task.get("payload") or {}
615
616 if action == "ping":
617 return {"message": "pong", "time": now_text()}
618
619 if action == "forbidden_scan":
620 # 兼容旧的扫描方式
621 return forbidden_scan_incremental(
622 root=payload["root"],
623 keywords=payload.get("keywords", []),
624 max_file_size=int(payload.get("max_file_size", 5 * 1024 * 1024)),
625 max_matches=int(payload.get("max_matches", 1000)),
626 scan_changed_only=payload.get("scan_changed_only", True),
627 scan_mode=payload.get("scan_mode", "incremental"),
628 )
629
630 raise ValueError("不支持的任务类型：" + str(action))
631
632
633def run_forbidden_scan(config):
634 """自动执行违禁词扫描任务"""
635 fb_config = get_forbidden_config(config)
636 if not fb_config or not fb_config.get("enabled"):
637 return
638
639 keywords = parse_keywords(fb_config.get("content", ""))
640 if not keywords:
641 log_info("违禁词列表为空，跳过扫描")
642 return
643
644 log_info("开始执行自动违禁词扫描")
645 init_index_db()
646 clean_orphaned_index(fb_config.get("scan_dir", "/www/wwwroot"))
647
648 task_id = f"scan_auto_{int(time.time())}"
649
650 try:
651 result = forbidden_scan_incremental(
652 root=fb_config.get("scan_dir", "/www/wwwroot"),
653 keywords=keywords,
654 max_file_size=fb_config.get("max_file_size", 5242880),
655 max_matches=fb_config.get("max_matches", 1000),
656 skip_dirs=parse_skip_dirs(fb_config.get("skip_dirs")),
657 skip_exts=parse_skip_exts(fb_config.get("skip_exts")),
658 scan_changed_only=fb_config.get("scan_changed_only", True),
659 scan_mode="incremental",
660 )
661 report_result(config, task_id, "success", result)
662 summary = result["summary"]
663 log_info("自动违禁词扫描完成：扫描文件 %d 个，命中 %d 条，耗时 %.1f 秒",
664 summary["scanned_files"], summary["matches"], summary.get("duration_sec", 0))
665 except Exception as e:
666 report_result(config, task_id, "failed", {}, str(e))
667 log_error("自动违禁词扫描失败：%s", e, exc_info=True)
668
669
670_last_full_scan_minute = None
671
672def should_run_full_scan(fb_config):
673 """判断是否应该进行全量扫描（同一分钟仅触发一次）"""
674 global _last_full_scan_minute
675 if not fb_config or not fb_config.get("full_scan_enabled"):
676 return False
677
678 cron_str = fb_config.get("full_scan_cron", "0 3 * * *")
679 try:
680 parts = cron_str.split()
681 if len(parts) >= 2:
682 minute, hour = parts[0], parts[1]
683 now = datetime.now()
684 if now.minute == int(minute) and now.hour == int(hour):
685 key = now.strftime("%Y%m%d%H%M")
686 if _last_full_scan_minute == key:
687 return False
688 _last_full_scan_minute = key
689 return True
690 except (ValueError, IndexError):
691 pass
692
693 return False
694
695
696def run_once(config):
697 """执行一次完整的工作周期"""
698 log_info("====== 开始工作周期 ======")
699 log_info("步骤1: heartbeat")
700 try:
701 heartbeat(config)
702 log_info("heartbeat 成功")
703 except Exception as exc:
704 log_warn("heartbeat 失败（继续）：%s", exc)
705
706 log_info("步骤2: collect_site_stats")
707 try:
708 collect_site_stats(config)
709 log_info("collect_site_stats 完成")
710 except Exception as exc:
711 log_error("collect_site_stats 异常：%s", exc, exc_info=True)
712
713 log_info("步骤3: get_forbidden_config")
714 fb_config = get_forbidden_config(config)
715 if fb_config and fb_config.get("enabled"):
716 log_info("违禁词扫描已启用，检查是否需要扫描")
717 if should_run_full_scan(fb_config):
718 log_info("执行全量扫描...")
719 run_forbidden_scan_with_mode(config, "full")
720 else:
721 log_info("执行增量扫描...")
722 run_forbidden_scan_with_mode(config, "incremental")
723 else:
724 log_info("违禁词扫描未启用或配置为空")
725
726 log_info("步骤4: pull_task")
727 try:
728 task = pull_task(config)
729 log_debug("pull_task 成功")
730 except Exception as exc:
731 log_warn("pull_task 失败（跳过）：%s", exc)
732 log_info("====== 工作周期结束 ======")
733 return
734 if not task:
735 log_info("无待执行任务")
736 log_info("====== 工作周期结束 ======")
737 return
738
739 task_id = task.get("task_id")
740 log_info("步骤5: 执行任务 %s (action=%s)", task_id, task.get("action"))
741 try:
742 result = execute_task(task, config)
743 report_result(config, task_id, "success", result)
744 log_info("任务 %s 执行成功", task_id)
745 except Exception as exc:
746 report_result(config, task_id, "failed", {}, str(exc))
747 log_error("任务 %s 执行失败: %s", task_id, exc, exc_info=True)
748
749 log_info("====== 工作周期结束 ======")
750
751
752def run_forbidden_scan_with_mode(config, scan_mode):
753 """按指定模式执行违禁词扫描"""
754 fb_config = get_forbidden_config(config)
755 if not fb_config or not fb_config.get("enabled"):
756 log_info("违禁词扫描未启用，跳过 %s 扫描", scan_mode)
757 return
758
759 keywords = parse_keywords(fb_config.get("content", ""))
760 if not keywords:
761 log_info("违禁词列表为空，跳过扫描")
762 return
763
764 log_info("开始%s违禁词扫描", "全量" if scan_mode == "full" else "增量")
765 init_index_db()
766 clean_orphaned_index(fb_config.get("scan_dir", "/www/wwwroot"))
767
768 task_id = f"scan_{scan_mode}_{int(time.time())}"
769
770 try:
771 result = forbidden_scan_incremental(
772 root=fb_config.get("scan_dir", "/www/wwwroot"),
773 keywords=keywords,
774 max_file_size=fb_config.get("max_file_size", 5242880),
775 max_matches=fb_config.get("max_matches", 1000),
776 skip_dirs=parse_skip_dirs(fb_config.get("skip_dirs")),
777 skip_exts=parse_skip_exts(fb_config.get("skip_exts")),
778 scan_changed_only=(scan_mode != "full"),
779 scan_mode=scan_mode,
780 )
781 report_result(config, task_id, "success", result)
782 summary = result["summary"]
783 log_info("%s扫描完成：扫描文件 %d 个，命中 %d 条，耗时 %.1f 秒",
784 "全量" if scan_mode == "full" else "增量",
785 summary["scanned_files"], summary["matches"],
786 summary.get("duration_sec", 0))
787 except Exception as e:
788 report_result(config, task_id, "failed", {}, str(e))
789 log_error("%s扫描失败：%s", "全量" if scan_mode == "full" else "增量", e, exc_info=True)
790
791
792def do_full_scan(config):
793 """执行一次全量扫描"""
794 fb_config = get_forbidden_config(config)
795 if not fb_config or not fb_config.get("enabled"):
796 log_warn("违禁词扫描未启用，无法执行全量扫描")
797 return
798 run_forbidden_scan_with_mode(config, "full")
799
800
801def main():
802 once = "--once" in sys.argv
803 full_scan = "--full-scan" in sys.argv
804
805 setup_worker_log()
806
807 log_info("====== worker.py 启动 ======")
808 log_info("参数: %s", sys.argv)
809
810 log_info("初始化索引数据库...")
811 init_index_db()
812 log_info("初始化站点统计数据库...")
813 init_stats_db()
814
815 log_info("加载配置...")
816 config = load_config()
817 log_info("配置加载完成, mnbt_url=%s, node_id=%s",
818 config.get("mnbt_url", ""), config.get("node_id", ""))
819
820 if full_scan:
821 log_info("执行全量扫描...")
822 do_full_scan(config)
823 log_info("全量扫描完成")
824 return
825
826 while True:
827 try:
828 run_once(config)
829 except (urllib.error.URLError, TimeoutError, ValueError, OSError, json.JSONDecodeError, KeyError) as exc:
830 log_error("工作进程异常：%s", exc, exc_info=True)
831 except Exception as exc:
832 log_error("工作进程未预期异常：%s", exc, exc_info=True)
833 if once:
834 break
835 time.sleep(int(config.get("interval_seconds", 10)))
836
837 log_info("====== worker.py 退出 ======")
838
839
840if __name__ == "__main__":
841 main()