Edu/scripts/arch-scan/scanners/py-scanner.ts

import path from "node:path";
import fs from "node:fs";
import type { Database as DBType } from "better-sqlite3";

interface ScanStats {
  modules: number;
  symbols: number;
}

/** 递归收集目录下匹配扩展名的所有文件 */
function walkDir(dir: string, exts: string[]): string[] {
  if (!fs.existsSync(dir)) return [];
  const results: string[] = [];
  const stack: string[] = [dir];
  while (stack.length > 0) {
    const current = stack.pop()!;
    let entries: fs.Dirent[];
    try {
      entries = fs.readdirSync(current, { withFileTypes: true });
    } catch {
      continue;
    }
    for (const entry of entries) {
      const fullPath = path.join(current, entry.name);
      if (entry.isDirectory()) {
        if (
          [
            "node_modules",
            "dist",
            ".git",
            "__pycache__",
            ".venv",
            "venv",
          ].includes(entry.name)
        )
          continue;
        stack.push(fullPath);
      } else if (entry.isFile()) {
        const ext = path.extname(entry.name);
        if (exts.includes(ext)) results.push(fullPath);
      }
    }
  }
  return results;
}

// Python 符号正则
const PY_FUNC_RE = /^(?:async\s+)?def\s+(\w+)\s*\(/gm;
const PY_CLASS_RE = /^class\s+(\w+)\s*[\(:]/gm;

export function scanPython(db: DBType, root: string): ScanStats {
  const insertModule = db.prepare(
    "INSERT OR IGNORE INTO modules (name, path, language, service, type) VALUES (?, ?, ?, ?, ?)",
  );
  const insertSymbol = db.prepare(
    "INSERT INTO symbols (module_id, name, kind, language, file_path, line_start, line_end, is_exported) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
  );
  const getModuleId = db.prepare("SELECT id FROM modules WHERE name = ?") as {
    get: (name: string) => { id: number } | undefined;
  };

  let modules = 0;
  let symbols = 0;

  // 扫描 services/* 下有 pyproject.toml 的服务
  const servicesDir = path.join(root, "services");
  if (!fs.existsSync(servicesDir)) return { modules, symbols };

  const entries = fs.readdirSync(servicesDir, { withFileTypes: true });
  for (const entry of entries) {
    if (!entry.isDirectory()) continue;
    const serviceName = entry.name;
    const servicePath = path.join(servicesDir, serviceName);
    const hasPyproject = fs.existsSync(
      path.join(servicePath, "pyproject.toml"),
    );
    if (!hasPyproject) continue;

    insertModule.run(
      serviceName,
      servicePath,
      "python",
      serviceName,
      "service",
    );
    const modRow = getModuleId.get(serviceName);
    if (modRow) {
      modules++;
      // 扫描 .py 文件
      const pyFiles = walkDir(servicePath, [".py"]);
      for (const filePath of pyFiles) {
        let content: string;
        try {
          content = fs.readFileSync(filePath, "utf-8");
        } catch {
          continue;
        }

        // 函数（含 async def）
        let match: RegExpExecArray | null;
        const funcRe = new RegExp(PY_FUNC_RE);
        while ((match = funcRe.exec(content)) !== null) {
          const name = match[1];
          const lineNum = content.slice(0, match.index).split("\n").length;
          // Python 中以 _ 开头的为私有
          const isExported = !name.startsWith("_") ? 1 : 0;
          insertSymbol.run(
            modRow.id,
            name,
            "function",
            "python",
            filePath,
            lineNum,
            lineNum,
            isExported,
          );
          symbols++;
        }

        // 类
        const classRe = new RegExp(PY_CLASS_RE);
        while ((match = classRe.exec(content)) !== null) {
          const name = match[1];
          const lineNum = content.slice(0, match.index).split("\n").length;
          insertSymbol.run(
            modRow.id,
            name,
            "class",
            "python",
            filePath,
            lineNum,
            lineNum,
            1,
          );
          symbols++;
        }
      }
    }
  }

  return { modules, symbols };
}