From 19930f56ce2a02be1d91580548204bd8bc8af171 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 13 Jun 2026 14:29:08 +0000 Subject: [PATCH] feat: add personal wechat wcf collector deploy --- config/platform-infra/wechat-archive.yaml | 10 + scripts/src/help.ts | 2 + scripts/src/platform-infra-wechat-archive.ts | 858 +++++++++++++++++- scripts/src/platform-infra.ts | 5 + scripts/windows/personal-wechat/prepare.ps1 | 115 +++ scripts/windows/personal-wechat/start.ps1 | 30 + scripts/windows/personal-wechat/status.ps1 | 54 ++ scripts/windows/personal-wechat/wcf_host.py | 76 ++ .../personal-wechat-collector/Dockerfile | 18 + .../personal-wechat-collector/collector.py | 282 ++++++ 10 files changed, 1447 insertions(+), 3 deletions(-) create mode 100644 scripts/windows/personal-wechat/prepare.ps1 create mode 100644 scripts/windows/personal-wechat/start.ps1 create mode 100644 scripts/windows/personal-wechat/status.ps1 create mode 100644 scripts/windows/personal-wechat/wcf_host.py create mode 100644 src/components/platform-infra/personal-wechat-collector/Dockerfile create mode 100644 src/components/platform-infra/personal-wechat-collector/collector.py diff --git a/config/platform-infra/wechat-archive.yaml b/config/platform-infra/wechat-archive.yaml index 8fcf61d1..e4ba750f 100644 --- a/config/platform-infra/wechat-archive.yaml +++ b/config/platform-infra/wechat-archive.yaml @@ -63,6 +63,7 @@ personalWechatIngress: isolation: dedicated-install requiredVersion: 3.9.12.51 installerAsset: WeChatSetup-3.9.12.51.exe + installMode: silent-nsis-with-ui-fallback installRoot: C:/UniDesk/personal-wechat/wechat-3.9.12.51 dataRoot: C:/UniDesk/personal-wechat/wechat-data autoUpdatePolicy: disabled @@ -97,7 +98,15 @@ personalWechatIngress: kind: PersistentVolumeClaim name: personal-wechat-collector-state mountPath: /var/lib/unidesk/personal-wechat + size: 1Gi create: true + image: + repository: 127.0.0.1:5000/platform-infra/personal-wechat-collector + tag: wcferry-39.5.2.0 + pullPolicy: IfNotPresent + context: src/components/platform-infra/personal-wechat-collector + dockerfile: src/components/platform-infra/personal-wechat-collector/Dockerfile + wcferryVersion: 39.5.2.0 wcfHost: 172.26.16.1 commandPort: 10086 messagePort: 10087 @@ -105,6 +114,7 @@ personalWechatIngress: queueMode: sqlite-wal outboxMode: archive-callback archiveCallbackRef: archiveCallback + secretName: personal-wechat-collector-secrets readOnly: enabled: true sendCapability: false diff --git a/scripts/src/help.ts b/scripts/src/help.ts index c7592971..b92ada5f 100644 --- a/scripts/src/help.ts +++ b/scripts/src/help.ts @@ -629,6 +629,8 @@ function platformInfraHelpSummary(): unknown { "bun scripts/cli.ts platform-infra wechat-archive apply --confirm", "bun scripts/cli.ts platform-infra wechat-archive validate --full", "bun scripts/cli.ts platform-infra wechat-archive pull --remote-path /UniDesk/WeChatArchive/...", + "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full", + "bun scripts/cli.ts platform-infra wechat-archive collector-status --full", ], description: "Operate G14 platform-infra services such as Sub2API, LangBot, n8n, WeChat archive workflows, and the YAML-controlled Codex pool.", }; diff --git a/scripts/src/platform-infra-wechat-archive.ts b/scripts/src/platform-infra-wechat-archive.ts index 7de3bd40..90ef4392 100644 --- a/scripts/src/platform-infra-wechat-archive.ts +++ b/scripts/src/platform-infra-wechat-archive.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { Buffer } from "node:buffer"; import { existsSync, readFileSync } from "node:fs"; import pathPosix from "node:path/posix"; import type { UniDeskConfig } from "./config"; @@ -33,7 +34,7 @@ import { type OpsCommonOptions, } from "./platform-infra-ops-library"; import { prepareLangBotSecretMaterial, readLangBotPostgresConninfo, readLangBotRuntimeConfig, readLangBotSecretMaterial } from "./platform-infra-langbot"; -import { fingerprintValues } from "./platform-infra-public-service"; +import { capture, compactCapture, fingerprintValues, parseJsonOutput, shQuote } from "./platform-infra-public-service"; const configFile = rootPath("config", "platform-infra", "wechat-archive.yaml"); const configLabel = "config/platform-infra/wechat-archive.yaml"; @@ -72,6 +73,7 @@ interface WechatArchiveConfig { isolation: string; requiredVersion: string; installerAsset: string; + installMode: string; installRoot: string; dataRoot: string; autoUpdatePolicy: string; @@ -94,7 +96,8 @@ interface WechatArchiveConfig { namespace: string; createNamespace: boolean; workload: { kind: string; name: string; serviceAccountName: string; replicas: number; containerName: string }; - storage: { kind: string; name: string; mountPath: string; create: boolean }; + storage: { kind: string; name: string; mountPath: string; size: string; create: boolean }; + image: { repository: string; tag: string; pullPolicy: string; context: string; dockerfile: string; wcferryVersion: string }; wcfHost: string; commandPort: number; messagePort: number; @@ -102,6 +105,7 @@ interface WechatArchiveConfig { queueMode: string; outboxMode: string; archiveCallbackRef: string; + secretName: string; readOnly: { enabled: boolean; sendCapability: boolean; allowedMethods: string[]; forbiddenMethods: string[] }; }; poc: { accountPolicy: string; observationWindowHours: number; requiredMessageTypes: string[] }; @@ -139,7 +143,7 @@ interface PullOptions extends OpsCommonOptions { export function wechatArchiveHelp(): Record { return { - command: "platform-infra wechat-archive plan|apply|status|validate|pull", + command: "platform-infra wechat-archive plan|apply|status|validate|pull|wcf-host-*|collector-*", output: "json", usage: [ "bun scripts/cli.ts platform-infra wechat-archive plan [--target G14]", @@ -149,6 +153,14 @@ export function wechatArchiveHelp(): Record { "bun scripts/cli.ts platform-infra wechat-archive validate [--target G14] [--full|--raw]", "bun scripts/cli.ts platform-infra wechat-archive pull --remote-path /UniDesk/WeChatArchive/... [--target G14] [--full|--raw]", "bun scripts/cli.ts platform-infra wechat-archive pull --fs-id [--local-path wechat-archive/pulls/file] [--target G14]", + "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm", + "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status [--full|--raw]", + "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm", + "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm", + "bun scripts/cli.ts platform-infra wechat-archive collector-image-status [--full|--raw]", + "bun scripts/cli.ts platform-infra wechat-archive collector-apply --dry-run", + "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm", + "bun scripts/cli.ts platform-infra wechat-archive collector-status [--full|--raw]", ], configTruth: configLabel, boundary: { @@ -168,6 +180,14 @@ export async function runWechatArchiveCommand(config: UniDeskConfig, args: strin if (action === "status") return await status(parseOpsCommonOptions(args.slice(1))); if (action === "validate") return await validate(parseOpsCommonOptions(args.slice(1))); if (action === "pull") return await pull(parsePullOptions(args.slice(1))); + if (action === "wcf-host-prepare") return await wcfHostPrepare(config, parseOpsApplyOptions(args.slice(1))); + if (action === "wcf-host-start") return await wcfHostStart(config, parseOpsApplyOptions(args.slice(1))); + if (action === "wcf-host-status") return await wcfHostStatus(config, parseOpsCommonOptions(args.slice(1))); + if (action === "collector-plan") return collectorPlan(parseOpsCommonOptions(args.slice(1))); + if (action === "collector-image-build") return await collectorImageBuild(config, parseOpsApplyOptions(args.slice(1))); + if (action === "collector-image-status") return await collectorImageStatus(config, parseOpsCommonOptions(args.slice(1))); + if (action === "collector-apply") return await collectorApply(config, parseOpsApplyOptions(args.slice(1))); + if (action === "collector-status") return await collectorStatus(config, parseOpsCommonOptions(args.slice(1))); return { ok: false, error: "unsupported-platform-infra-wechat-archive-command", args, help: wechatArchiveHelp() }; } @@ -206,6 +226,13 @@ function plan(options: OpsCommonOptions): Record { apply: `bun scripts/cli.ts platform-infra wechat-archive apply --target ${archive.target.id} --confirm`, validate: `bun scripts/cli.ts platform-infra wechat-archive validate --target ${archive.target.id} --full`, pull: `bun scripts/cli.ts platform-infra wechat-archive pull --remote-path ${archive.baiduNetdisk.archive.remoteRoot}/... --target ${archive.target.id}`, + personalWechatIngress: { + prepareWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm", + startWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm", + buildCollectorImage: "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm", + applyCollector: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm", + status: "bun scripts/cli.ts platform-infra wechat-archive collector-status --full", + }, }, }; } @@ -360,6 +387,246 @@ async function pull(options: PullOptions): Promise> { }; } +function collectorPlan(options: OpsCommonOptions): Record { + const archive = readConfig(); + assertTarget(archive, options.targetId); + const manifest = renderCollectorManifest(archive); + const policy = collectorPolicyChecks(archive, manifest); + return { + ok: policy.every((check) => check.ok), + action: "platform-infra-wechat-archive-collector-plan", + mutation: false, + config: personalWechatIngressSummary(archive.personalWechatIngress), + manifest: { + namespace: archive.personalWechatIngress.collector.namespace, + image: collectorImageRef(archive.personalWechatIngress), + bytes: Buffer.byteLength(manifest, "utf8"), + sha256: fingerprintValues({ manifest }, ["manifest"]), + }, + policy, + next: { + prepareWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm", + startWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm", + buildImage: "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm", + dryRun: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --dry-run", + apply: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm", + status: "bun scripts/cli.ts platform-infra wechat-archive collector-status --full", + }, + }; +} + +async function wcfHostPrepare(config: UniDeskConfig, options: OpsApplyOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + if (options.dryRun) { + return { + ok: true, + action: "platform-infra-wechat-archive-wcf-host-prepare", + mode: "dry-run", + mutation: false, + target: archive.personalWechatIngress.target, + scripts: windowsScriptSummary(), + }; + } + const result = await capture(config, archive.personalWechatIngress.target.windowsRoute, ["ps"], windowsScriptSyncAndStartPrepareScript(archive)); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && boolField(parsed, "ok", false), + action: "platform-infra-wechat-archive-wcf-host-prepare", + mode: "confirmed", + mutation: true, + target: archive.personalWechatIngress.target, + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + next: { + status: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full", + start: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm", + }, + }; +} + +async function wcfHostStart(config: UniDeskConfig, options: OpsApplyOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + if (options.dryRun) { + return { + ok: true, + action: "platform-infra-wechat-archive-wcf-host-start", + mode: "dry-run", + mutation: false, + target: archive.personalWechatIngress.target, + }; + } + const result = await capture(config, archive.personalWechatIngress.target.windowsRoute, ["ps"], windowsRunManagedScript(archive, "start.ps1")); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && boolField(parsed, "ok", false), + action: "platform-infra-wechat-archive-wcf-host-start", + mode: "confirmed", + mutation: true, + target: archive.personalWechatIngress.target, + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + next: { + status: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full", + collectorApply: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm", + }, + }; +} + +async function wcfHostStatus(config: UniDeskConfig, options: OpsCommonOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + const result = await capture(config, archive.personalWechatIngress.target.windowsRoute, ["ps"], windowsRunManagedScript(archive, "status.ps1")); + const parsed = parseJsonOutput(result.stdout); + const status = parsed === null ? null : compactWcfHostStatus(parsed, options.full); + return { + ok: result.exitCode === 0 && parsed !== null && parsed.ok === true, + action: "platform-infra-wechat-archive-wcf-host-status", + target: archive.personalWechatIngress.target, + summary: status, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + }; +} + +async function collectorImageBuild(config: UniDeskConfig, options: OpsApplyOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + if (options.confirm && !options.wait) { + const job = startJob( + "platform_infra_wechat_archive_collector_image_build_d601", + ["bun", "scripts/cli.ts", "platform-infra", "wechat-archive", "collector-image-build", "--confirm", "--wait"], + "Build and push the D601 k3s personal WeChat collector image through the controlled UniDesk CLI", + ); + return { + ok: true, + action: "platform-infra-wechat-archive-collector-image-build", + mode: "async-job", + mutation: true, + job, + statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`, + }; + } + if (options.dryRun) { + return { + ok: true, + action: "platform-infra-wechat-archive-collector-image-build", + mode: "dry-run", + mutation: false, + image: collectorImageRef(archive.personalWechatIngress), + source: collectorSourceSummary(archive), + }; + } + const result = await capture(config, archive.personalWechatIngress.target.hostRoute, ["script"], collectorImageBuildScript(archive)); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && boolField(parsed, "ok", false), + action: "platform-infra-wechat-archive-collector-image-build", + mode: "confirmed", + mutation: true, + image: collectorImageRef(archive.personalWechatIngress), + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + }; +} + +async function collectorImageStatus(config: UniDeskConfig, options: OpsCommonOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + const result = await capture(config, archive.personalWechatIngress.target.hostRoute, ["script"], collectorImageStatusScript(archive)); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && parsed !== null && parsed.ok === true, + action: "platform-infra-wechat-archive-collector-image-status", + image: collectorImageRef(archive.personalWechatIngress), + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + }; +} + +async function collectorApply(config: UniDeskConfig, options: OpsApplyOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + const manifest = renderCollectorManifest(archive); + const policy = collectorPolicyChecks(archive, manifest); + if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-wechat-archive-collector-apply", mode: "policy-blocked", policy }; + if (options.confirm && !options.wait) { + const job = startJob( + "platform_infra_wechat_archive_collector_apply_d601", + ["bun", "scripts/cli.ts", "platform-infra", "wechat-archive", "collector-apply", "--confirm", "--wait"], + "Apply D601 k3s personal WeChat collector objects through the controlled UniDesk CLI", + ); + return { + ok: true, + action: "platform-infra-wechat-archive-collector-apply", + mode: "async-job", + mutation: true, + job, + statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`, + }; + } + if (options.dryRun) { + const result = await capture(config, archive.personalWechatIngress.target.kubeRoute, ["script"], collectorApplyScript(archive, manifest, null, true)); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && boolField(parsed, "ok", false), + action: "platform-infra-wechat-archive-collector-apply", + mode: "dry-run", + mutation: false, + policy, + target: archive.personalWechatIngress.target, + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + }; + } + const token = readArchiveCallbackToken(archive); + const result = await capture(config, archive.personalWechatIngress.target.kubeRoute, ["script"], collectorApplyScript(archive, manifest, token.value, false)); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && boolField(parsed, "ok", false), + action: "platform-infra-wechat-archive-collector-apply", + mode: "confirmed", + mutation: true, + policy, + target: archive.personalWechatIngress.target, + secret: { + sourceRef: token.sourceRef, + keyName: token.keyName, + targetSecret: archive.personalWechatIngress.collector.secretName, + fingerprint: token.fingerprint, + valuesPrinted: false, + }, + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + next: { + status: "bun scripts/cli.ts platform-infra wechat-archive collector-status --full", + wcfHostStatus: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full", + }, + }; +} + +async function collectorStatus(config: UniDeskConfig, options: OpsCommonOptions): Promise> { + const archive = readConfig(); + assertTarget(archive, options.targetId); + const result = await capture(config, archive.personalWechatIngress.target.kubeRoute, ["script"], collectorStatusScript(archive, options.full)); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && parsed !== null && parsed.ok === true, + action: "platform-infra-wechat-archive-collector-status", + target: archive.personalWechatIngress.target, + summary: parsed, + remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }), + ...(options.raw ? { raw: result } : {}), + }; +} + function readConfig(): WechatArchiveConfig { const root = readYamlRecord(configFile, "platform-infra-wechat-archive"); const metadata = recordField(root, "metadata", configLabel); @@ -475,6 +742,562 @@ function readConfig(): WechatArchiveConfig { return config; } +const collectorFieldManager = "unidesk-platform-infra-wechat-archive"; +const windowsOpsRoot = "C:\\UniDesk\\personal-wechat\\ops"; + +function collectorImageRef(config: WechatArchiveConfig["personalWechatIngress"]): string { + return `${config.collector.image.repository}:${config.collector.image.tag}`; +} + +function boolField(value: Record | null, key: string, fallback: boolean): boolean { + if (value === null) return fallback; + const raw = value[key]; + return typeof raw === "boolean" ? raw : fallback; +} + +function collectorSourceSummary(archive: WechatArchiveConfig): Record { + const ingress = archive.personalWechatIngress; + const dockerfile = resolveRepoPath(ingress.collector.image.dockerfile); + const collectorPy = resolveRepoPath(pathPosix.join(ingress.collector.image.context, "collector.py")); + return { + dockerfile: ingress.collector.image.dockerfile, + collectorPy: repoRelative(collectorPy), + dockerfileSha256: sha256File(dockerfile), + collectorPySha256: sha256File(collectorPy), + wcferryVersion: ingress.collector.image.wcferryVersion, + }; +} + +function windowsScriptSummary(): Record { + const files = ["prepare.ps1", "start.ps1", "status.ps1", "wcf_host.py"].map((name) => { + const filePath = rootPath("scripts", "windows", "personal-wechat", name); + return { name, path: repoRelative(filePath), sha256: sha256File(filePath), bytes: readFileSync(filePath).byteLength }; + }); + return { targetRoot: windowsOpsRoot, files }; +} + +function b64(value: string): string { + return Buffer.from(value, "utf8").toString("base64"); +} + +function fileB64(repoPath: string): string { + return readFileSync(resolveRepoPath(repoPath)).toString("base64"); +} + +function managedWindowsFiles(): Array<{ name: string; contentB64: string }> { + const base = rootPath("scripts", "windows", "personal-wechat"); + return ["prepare.ps1", "start.ps1", "status.ps1", "wcf_host.py"].map((name) => ({ + name, + contentB64: readFileSync(pathPosix.join(base, name)).toString("base64"), + })); +} + +function windowsScriptSyncBlock(): string { + const writes = managedWindowsFiles().map((file) => { + const target = `${windowsOpsRoot}\\${file.name}`; + return ` +$target = ${psSingleQuote(target)} +$content = [Text.Encoding]::UTF8.GetString([Convert]::FromBase64String(${psSingleQuote(file.contentB64)})) +Set-Content -LiteralPath $target -Encoding UTF8 -Value $content +`; + }).join("\n"); + return ` +$ErrorActionPreference = "Stop" +$OpsRoot = ${psSingleQuote(windowsOpsRoot)} +New-Item -ItemType Directory -Force $OpsRoot | Out-Null +${writes} +`; +} + +function windowsScriptSyncAndStartPrepareScript(_archive: WechatArchiveConfig): string { + const prepare = `${windowsOpsRoot}\\prepare.ps1`; + const stateRoot = "C:\\UniDesk\\personal-wechat\\wcf-state"; + const stdout = `${stateRoot}\\prepare.stdout.log`; + const stderr = `${stateRoot}\\prepare.stderr.log`; + const resultFile = `${stateRoot}\\prepare-result.json`; + return `${windowsScriptSyncBlock()} +$StateRoot = ${psSingleQuote(stateRoot)} +New-Item -ItemType Directory -Force $StateRoot | Out-Null +if (Test-Path ${psSingleQuote(resultFile)}) { Remove-Item -Force ${psSingleQuote(resultFile)} } +$proc = Start-Process -FilePath "powershell.exe" -ArgumentList @("-NoProfile","-ExecutionPolicy","Bypass","-File",${psSingleQuote(prepare)}) -WindowStyle Hidden -PassThru -RedirectStandardOutput ${psSingleQuote(stdout)} -RedirectStandardError ${psSingleQuote(stderr)} +[pscustomobject]@{ + ok = $true + started = $true + pid = $proc.Id + opsRoot = $OpsRoot + stdout = ${psSingleQuote(stdout)} + stderr = ${psSingleQuote(stderr)} + result = ${psSingleQuote(resultFile)} + statusCommand = "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full" +} | ConvertTo-Json -Depth 6 +`; +} + +function windowsRunManagedScript(_archive: WechatArchiveConfig, scriptName: "start.ps1" | "status.ps1"): string { + const script = `${windowsOpsRoot}\\${scriptName}`; + return `${windowsScriptSyncBlock()} +& ${psSingleQuote(script)} +if ($LASTEXITCODE -ne $null -and $LASTEXITCODE -ne 0) { exit $LASTEXITCODE } +`; +} + +function psSingleQuote(value: string): string { + return `'${value.replaceAll("'", "''")}'`; +} + +function compactWcfHostStatus(parsed: Record, full: boolean): Record { + const status = typeof parsed.status === "object" && parsed.status !== null && !Array.isArray(parsed.status) + ? parsed.status as Record + : null; + const prepare = typeof parsed.prepare === "object" && parsed.prepare !== null && !Array.isArray(parsed.prepare) + ? parsed.prepare as Record + : null; + return { + ok: parsed.ok, + prepared: parsed.prepared, + running: parsed.running, + pid: parsed.pid, + ports: parsed.ports, + login: status === null ? null : { + isLogin: status.isLogin, + qrcodePresent: status.qrcodePresent, + qrcode: full ? status.qrcode : undefined, + ts: status.ts, + }, + prepare: prepare === null ? null : { + ok: prepare.ok, + wechatExe: prepare.wechatExe, + installAttempted: prepare.installAttempted, + installExitCode: prepare.installExitCode, + wcferryVersion: prepare.wcferryVersion, + next: prepare.next, + }, + paths: parsed.paths, + logs: full ? parsed.logs : undefined, + valuesPrinted: false, + }; +} + +function renderCollectorManifest(archive: WechatArchiveConfig): string { + const ingress = archive.personalWechatIngress; + const collector = ingress.collector; + const labels = [ + "app.kubernetes.io/name: personal-wechat-collector", + "app.kubernetes.io/part-of: platform-infra", + "app.kubernetes.io/managed-by: unidesk", + "unidesk.ai/component: personal-wechat-collector", + ].join("\n "); + return `apiVersion: v1 +kind: ServiceAccount +metadata: + name: ${collector.workload.serviceAccountName} + namespace: ${collector.namespace} + labels: + ${labels} +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: ${collector.storage.name} + namespace: ${collector.namespace} + labels: + ${labels} +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: ${collector.storage.size} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: personal-wechat-collector-config + namespace: ${collector.namespace} + labels: + ${labels} +data: + WCF_HOST: "${collector.wcfHost}" + WCF_COMMAND_PORT: "${collector.commandPort}" + STATE_ROOT: "${collector.stateRoot}" + ARCHIVE_CALLBACK_URL: "${archive.archiveCallback.publicUrl}" + ARCHIVE_REMOTE_ROOT: "${archive.baiduNetdisk.archive.remoteRoot}" + ARCHIVE_TIMEZONE: "${archive.baiduNetdisk.archive.timezone}" + POLL_IDLE_SECONDS: "1" + STATUS_INTERVAL_SECONDS: "15" +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: ${collector.workload.name} + namespace: ${collector.namespace} + labels: + ${labels} +spec: + replicas: ${collector.workload.replicas} + selector: + matchLabels: + app.kubernetes.io/name: personal-wechat-collector + unidesk.ai/component: personal-wechat-collector + template: + metadata: + labels: + app.kubernetes.io/name: personal-wechat-collector + app.kubernetes.io/part-of: platform-infra + unidesk.ai/component: personal-wechat-collector + annotations: + unidesk.ai/wcf-host: "${collector.wcfHost}:${collector.commandPort}" + unidesk.ai/read-only: "${collector.readOnly.enabled}" + unidesk.ai/send-capability: "${collector.readOnly.sendCapability}" + unidesk.ai/archive-callback-ref: "${collector.archiveCallbackRef}" + spec: + serviceAccountName: ${collector.workload.serviceAccountName} + containers: + - name: ${collector.workload.containerName} + image: ${collectorImageRef(ingress)} + imagePullPolicy: ${collector.image.pullPolicy} + envFrom: + - configMapRef: + name: personal-wechat-collector-config + env: + - name: ARCHIVE_CALLBACK_TOKEN + valueFrom: + secretKeyRef: + name: ${collector.secretName} + key: ${archive.archiveCallback.tokenKey} + volumeMounts: + - name: state + mountPath: ${collector.storage.mountPath} + volumes: + - name: state + persistentVolumeClaim: + claimName: ${collector.storage.name} +`; +} + +function collectorPolicyChecks(archive: WechatArchiveConfig, manifest: string): Array> { + const collector = archive.personalWechatIngress.collector; + return [ + { name: "no-namespace-create", ok: !/^\s*kind:\s*Namespace\s*$/mu.test(manifest) && collector.createNamespace === false, detail: "collector apply must reuse the existing platform-infra namespace and must not render/create Namespace." }, + { name: "target-d601-k3s", ok: archive.personalWechatIngress.target.kubeRoute === "D601:k3s" && collector.runtime === "d601-k3s", detail: "collector runtime is D601 native k3s." }, + { name: "namespace-platform-infra", ok: collector.namespace === "platform-infra", detail: "collector deploys into the existing platform-infra namespace." }, + { name: "no-public-exposure", ok: !/^\s*kind:\s*(Ingress|Service)\s*$/mu.test(manifest) && archive.personalWechatIngress.wcfHost.firewall.publicExposure === false, detail: "collector has no public ingress, node port, load balancer or service." }, + { name: "no-host-network", ok: !/^\s*hostNetwork:\s*true\s*$/mu.test(manifest), detail: "collector must not use hostNetwork." }, + { name: "read-only-methods", ok: collector.readOnly.enabled === true && collector.readOnly.sendCapability === false, detail: "collector only invokes read-only WCF methods in source." }, + { name: "callback-token-source-ref", ok: Boolean(archive.archiveCallback.tokenSourceRef && archive.archiveCallback.tokenKey && collector.secretName), detail: "callback token is synced from YAML-declared sourceRef into a runtime Secret." }, + ]; +} + +function collectorImageBuildScript(archive: WechatArchiveConfig): string { + const ingress = archive.personalWechatIngress; + const dockerfileB64 = fileB64(ingress.collector.image.dockerfile); + const collectorB64 = fileB64(pathPosix.join(ingress.collector.image.context, "collector.py")); + const image = collectorImageRef(ingress); + const jobRoot = "/tmp/unidesk-personal-wechat-collector-build"; + return ` +set -u +job_root=${shQuote(jobRoot)} +job_dir="$job_root/work" +status_file="$job_root/status.json" +runner="$job_root/build-runner.sh" +mkdir -p "$job_root" +if [ -f "$job_root/pid" ] && kill -0 "$(cat "$job_root/pid")" 2>/dev/null; then + python3 - "$status_file" "$(cat "$job_root/pid")" <<'PY' +import json, sys +payload = {"ok": True, "started": False, "alreadyRunning": True, "pid": sys.argv[2], "statusFile": sys.argv[1]} +print(json.dumps(payload, ensure_ascii=False, indent=2)) +PY + exit 0 +fi +rm -rf "$job_dir" +mkdir -p "$job_dir" +printf '%s' '${dockerfileB64}' | base64 -d > "$job_dir/Dockerfile" +printf '%s' '${collectorB64}' | base64 -d > "$job_dir/collector.py" +build_log="$job_dir/docker-build.log" +push_log="$job_dir/docker-push.log" +cat >"$runner" <<'SH' +#!/usr/bin/env bash +set +e +job_dir=${shQuote(`${jobRoot}/work`)} +status_file=${shQuote(`${jobRoot}/status.json`)} +build_log="$job_dir/docker-build.log" +push_log="$job_dir/docker-push.log" +python3 - "$status_file" "$build_log" "$push_log" <<'PY' +import json, sys, time +payload = {"ok": False, "status": "running", "startedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "image": "${image}", "logs": {"build": sys.argv[2], "push": sys.argv[3]}} +open(sys.argv[1], "w", encoding="utf-8").write(json.dumps(payload, ensure_ascii=False, indent=2)) +PY +docker build --build-arg WCFERRY_VERSION=${shQuote(ingress.collector.image.wcferryVersion)} -t ${shQuote(image)} "$job_dir" >"$build_log" 2>&1 +build_rc=$? +if [ "$build_rc" -eq 0 ]; then + docker push ${shQuote(image)} >"$push_log" 2>&1 + push_rc=$? +else + : >"$push_log" + push_rc=1 +fi +python3 - "$status_file" "$build_rc" "$push_rc" "$build_log" "$push_log" <<'PY' +import json, sys +status_file, build_rc, push_rc = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]) +def text(path, limit=8000): + try: + return open(path, encoding="utf-8", errors="replace").read()[-limit:] + except FileNotFoundError: + return "" +payload = { + "ok": build_rc == 0 and push_rc == 0, + "status": "succeeded" if build_rc == 0 and push_rc == 0 else "failed", + "image": "${image}", + "wcferryVersion": "${ingress.collector.image.wcferryVersion}", + "steps": { + "build": {"exitCode": build_rc, "log": sys.argv[4], "logTail": text(sys.argv[4])}, + "push": {"exitCode": push_rc, "log": sys.argv[5], "logTail": text(sys.argv[5])}, + }, +} +open(status_file, "w", encoding="utf-8").write(json.dumps(payload, ensure_ascii=False, indent=2) + "\\n") +sys.exit(0 if payload["ok"] else 1) +PY +SH +chmod +x "$runner" +nohup "$runner" >"$job_root/nohup.out" 2>"$job_root/nohup.err" & +pid=$! +printf '%s' "$pid" >"$job_root/pid" +python3 - "$status_file" "$pid" "$job_root" <<'PY' +import json, sys +payload = {"ok": True, "started": True, "pid": sys.argv[2], "statusFile": sys.argv[1], "jobRoot": sys.argv[3], "statusCommand": "bun scripts/cli.ts platform-infra wechat-archive collector-image-status --full"} +print(json.dumps(payload, ensure_ascii=False, indent=2)) +PY +`; +} + +function collectorImageStatusScript(archive: WechatArchiveConfig): string { + const image = collectorImageRef(archive.personalWechatIngress); + return ` +set -u +tmp="$(mktemp -d)" +trap 'rm -rf "$tmp"' EXIT +job_root=/tmp/unidesk-personal-wechat-collector-build +status_file="$job_root/status.json" +pid_file="$job_root/pid" +running=0 +if [ -f "$pid_file" ] && kill -0 "$(cat "$pid_file")" 2>/dev/null; then + running=1 +fi +docker image inspect ${shQuote(image)} >"$tmp/local.json" 2>"$tmp/local.err" +local_rc=$? +docker manifest inspect ${shQuote(image)} >"$tmp/manifest.json" 2>"$tmp/manifest.err" +manifest_rc=$? +python3 - "$local_rc" "$manifest_rc" "$running" "$status_file" "$tmp/local.json" "$tmp/local.err" "$tmp/manifest.json" "$tmp/manifest.err" <<'PY' +import json, sys +local_rc, manifest_rc, running = int(sys.argv[1]), int(sys.argv[2]), sys.argv[3] == "1" +def load(path): + try: + return json.load(open(path, encoding="utf-8")) + except Exception: + return None +def text(path): + try: + return open(path, encoding="utf-8", errors="replace").read()[-2000:] + except FileNotFoundError: + return "" +build_status = load(sys.argv[4]) +local = load(sys.argv[5]) +payload = { + "ok": True, + "image": "${image}", + "build": build_status, + "buildRunning": running, + "localPresent": local_rc == 0, + "registryManifestPresent": manifest_rc == 0, + "local": { + "id": local[0].get("Id") if isinstance(local, list) and local else None, + "created": local[0].get("Created") if isinstance(local, list) and local else None, + "repoDigests": local[0].get("RepoDigests") if isinstance(local, list) and local else [], + }, + "errors": { + "local": text(sys.argv[6]) if local_rc != 0 else "", + "manifest": text(sys.argv[8]) if manifest_rc != 0 else "", + }, +} +print(json.dumps(payload, ensure_ascii=False, indent=2)) +PY +`; +} + +function collectorApplyScript(archive: WechatArchiveConfig, manifest: string, token: string | null, dryRun: boolean): string { + const ingress = archive.personalWechatIngress; + const collector = ingress.collector; + const manifestB64 = b64(manifest); + const tokenB64 = token === null ? "" : b64(token); + const mode = dryRun ? "dry-run" : "confirmed"; + return ` +set -u +tmp="$(mktemp -d)" +trap 'rm -rf "$tmp"' EXIT +manifest="$tmp/personal-wechat-collector.yaml" +printf '%s' '${manifestB64}' | base64 -d > "$manifest" +kubectl get namespace ${collector.namespace} >"$tmp/ns.out" 2>"$tmp/ns.err" +ns_rc=$? +secret_rc=0 +apply_rc=1 +if [ "$ns_rc" -eq 0 ]; then + if [ ${dryRun ? "1" : "0"} -eq 1 ]; then + kubectl apply --dry-run=client -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err" + apply_rc=$? + else + printf '%s' '${tokenB64}' | base64 -d > "$tmp/archive-token" + kubectl -n ${collector.namespace} create secret generic ${collector.secretName} \\ + --from-file=${archive.archiveCallback.tokenKey}="$tmp/archive-token" \\ + --dry-run=client -o yaml | kubectl apply --server-side --force-conflicts --field-manager=${collectorFieldManager} -f - >"$tmp/secret.out" 2>"$tmp/secret.err" + secret_rc=$? + if [ "$secret_rc" -eq 0 ]; then + kubectl apply --server-side --force-conflicts --field-manager=${collectorFieldManager} -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err" + apply_rc=$? + else + : >"$tmp/apply.out" + printf '%s\\n' 'skipped because secret sync failed' >"$tmp/apply.err" + apply_rc=1 + fi + fi +else + : >"$tmp/secret.out"; : >"$tmp/secret.err"; : >"$tmp/apply.out" + printf '%s\\n' 'skipped because declared namespace does not exist' >"$tmp/apply.err" +fi +python3 - "$ns_rc" "$secret_rc" "$apply_rc" "$tmp/ns.out" "$tmp/ns.err" "$tmp/secret.out" "$tmp/secret.err" "$tmp/apply.out" "$tmp/apply.err" <<'PY' +import json, sys +ns_rc, secret_rc, apply_rc = int(sys.argv[1]), int(sys.argv[2]), int(sys.argv[3]) +def text(path, limit=8000): + try: + return open(path, encoding="utf-8", errors="replace").read()[-limit:] + except FileNotFoundError: + return "" +payload = { + "ok": ns_rc == 0 and secret_rc == 0 and apply_rc == 0, + "mode": "${mode}", + "namespace": "${collector.namespace}", + "image": "${collectorImageRef(ingress)}", + "secret": {"name": "${collector.secretName}", "key": "${archive.archiveCallback.tokenKey}", "applied": ${dryRun ? "False" : "True"}, "valuesPrinted": False}, + "steps": { + "namespace": {"exitCode": ns_rc, "stdout": text(sys.argv[4]), "stderr": text(sys.argv[5])}, + "secret": {"exitCode": secret_rc, "stdout": text(sys.argv[6]), "stderr": text(sys.argv[7])}, + "apply": {"exitCode": apply_rc, "stdout": text(sys.argv[8]), "stderr": text(sys.argv[9])}, + }, +} +print(json.dumps(payload, ensure_ascii=False, indent=2)) +sys.exit(0 if payload["ok"] else 1) +PY +`; +} + +function collectorStatusScript(archive: WechatArchiveConfig, full: boolean): string { + const collector = archive.personalWechatIngress.collector; + return ` +set -u +tmp="$(mktemp -d)" +trap 'rm -rf "$tmp"' EXIT +capture() { + name="$1" + shift + "$@" -o json >"$tmp/$name.json" 2>"$tmp/$name.err" + printf '%s' "$?" >"$tmp/$name.rc" +} +capture namespace kubectl get namespace ${collector.namespace} +capture deploy kubectl -n ${collector.namespace} get deploy ${collector.workload.name} +capture pods kubectl -n ${collector.namespace} get pods -l app.kubernetes.io/name=personal-wechat-collector +capture pvc kubectl -n ${collector.namespace} get pvc ${collector.storage.name} +capture secret kubectl -n ${collector.namespace} get secret ${collector.secretName} +capture configmap kubectl -n ${collector.namespace} get configmap personal-wechat-collector-config +capture events kubectl -n ${collector.namespace} get events --sort-by=.lastTimestamp +pod="$(kubectl -n ${collector.namespace} get pods -l app.kubernetes.io/name=personal-wechat-collector -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)" +if [ -n "$pod" ]; then + kubectl -n ${collector.namespace} logs "$pod" --tail ${full ? "160" : "40"} >"$tmp/logs.txt" 2>"$tmp/logs.err" + logs_rc=$? +else + : >"$tmp/logs.txt" + printf '%s\\n' 'collector pod not found' >"$tmp/logs.err" + logs_rc=1 +fi +python3 - "$tmp" "$logs_rc" "$pod" <<'PY' +import json, os, sys +tmp, logs_rc, pod = sys.argv[1], int(sys.argv[2]), sys.argv[3] +def load(name): + try: + return json.load(open(os.path.join(tmp, f"{name}.json"), encoding="utf-8")) + except Exception: + return None +def rc(name): + try: + return int(open(os.path.join(tmp, f"{name}.rc"), encoding="utf-8").read() or "1") + except FileNotFoundError: + return 1 +def text(path, limit=8000): + try: + return open(path, encoding="utf-8", errors="replace").read()[-limit:] + except FileNotFoundError: + return "" +deploy = load("deploy") or {} +status = deploy.get("status", {}) if isinstance(deploy, dict) else {} +spec = deploy.get("spec", {}) if isinstance(deploy, dict) else {} +pods = load("pods") or {"items": []} +events = load("events") or {"items": []} +pod_items = pods.get("items", []) if isinstance(pods, dict) else [] +def pod_summary(item): + pod_status = item.get("status", {}) + containers = pod_status.get("containerStatuses") or [] + return { + "name": item.get("metadata", {}).get("name"), + "phase": pod_status.get("phase"), + "podIP": pod_status.get("podIP"), + "nodeName": item.get("spec", {}).get("nodeName"), + "containers": [ + { + "name": c.get("name"), + "ready": c.get("ready"), + "restartCount": c.get("restartCount"), + "state": c.get("state"), + "lastState": c.get("lastState"), + } + for c in containers + ], + } +def event_summary(event): + involved = event.get("involvedObject", {}) + if involved.get("name") and "personal-wechat-collector" not in involved.get("name", ""): + return None + return { + "type": event.get("type"), + "reason": event.get("reason"), + "message": str(event.get("message") or "")[-600:], + "count": event.get("count"), + "lastTimestamp": event.get("lastTimestamp") or event.get("eventTime"), + "involvedObject": {"kind": involved.get("kind"), "name": involved.get("name")}, + } +related_events = [item for item in (event_summary(e) for e in events.get("items", [])[-30:]) if item] +ready = status.get("readyReplicas", 0) >= spec.get("replicas", 1) and rc("deploy") == 0 +payload = { + "ok": rc("namespace") == 0 and rc("deploy") == 0, + "namespace": "${collector.namespace}", + "deployment": { + "name": "${collector.workload.name}", + "replicas": spec.get("replicas"), + "readyReplicas": status.get("readyReplicas", 0), + "updatedReplicas": status.get("updatedReplicas", 0), + "availableReplicas": status.get("availableReplicas", 0), + "ready": ready, + }, + "pvc": {"present": rc("pvc") == 0, "name": "${collector.storage.name}"}, + "secret": {"present": rc("secret") == 0, "name": "${collector.secretName}", "valuesPrinted": False}, + "configMap": {"present": rc("configmap") == 0, "name": "personal-wechat-collector-config"}, + "pods": [pod_summary(item) for item in pod_items], + "logs": {"pod": pod, "exitCode": logs_rc, "tail": text(os.path.join(tmp, "logs.txt")), "stderrTail": text(os.path.join(tmp, "logs.err"), 2000)}, + "events": related_events[-12:], + "valuesPrinted": False, +} +print(json.dumps(payload, ensure_ascii=False, indent=2)) +PY +`; +} + function validateConfig(config: WechatArchiveConfig): void { if (config.version !== 1) throw new Error(`${configLabel}.version must be 1`); if (config.target.id !== "G14" || config.target.route !== "G14:k3s") throw new Error(`${configLabel}.target currently supports only G14:k3s`); @@ -575,6 +1398,7 @@ function parsePersonalWechatIngress(raw: Record): WechatArchive const collector = recordField(raw, "collector", `${configLabel}.personalWechatIngress`); const workload = recordField(collector, "workload", `${configLabel}.personalWechatIngress.collector`); const storage = recordField(collector, "storage", `${configLabel}.personalWechatIngress.collector`); + const image = recordField(collector, "image", `${configLabel}.personalWechatIngress.collector`); const readOnly = recordField(collector, "readOnly", `${configLabel}.personalWechatIngress.collector`); const poc = recordField(raw, "poc", `${configLabel}.personalWechatIngress`); return { @@ -590,6 +1414,7 @@ function parsePersonalWechatIngress(raw: Record): WechatArchive isolation: stringField(pcWechat, "isolation", `${configLabel}.personalWechatIngress.pcWechat`), requiredVersion: stringField(pcWechat, "requiredVersion", `${configLabel}.personalWechatIngress.pcWechat`), installerAsset: stringField(pcWechat, "installerAsset", `${configLabel}.personalWechatIngress.pcWechat`), + installMode: stringField(pcWechat, "installMode", `${configLabel}.personalWechatIngress.pcWechat`), installRoot: stringField(pcWechat, "installRoot", `${configLabel}.personalWechatIngress.pcWechat`), dataRoot: stringField(pcWechat, "dataRoot", `${configLabel}.personalWechatIngress.pcWechat`), autoUpdatePolicy: stringField(pcWechat, "autoUpdatePolicy", `${configLabel}.personalWechatIngress.pcWechat`), @@ -625,8 +1450,17 @@ function parsePersonalWechatIngress(raw: Record): WechatArchive kind: stringField(storage, "kind", `${configLabel}.personalWechatIngress.collector.storage`), name: stringField(storage, "name", `${configLabel}.personalWechatIngress.collector.storage`), mountPath: stringField(storage, "mountPath", `${configLabel}.personalWechatIngress.collector.storage`), + size: stringField(storage, "size", `${configLabel}.personalWechatIngress.collector.storage`), create: booleanField(storage, "create", `${configLabel}.personalWechatIngress.collector.storage`), }, + image: { + repository: stringField(image, "repository", `${configLabel}.personalWechatIngress.collector.image`), + tag: stringField(image, "tag", `${configLabel}.personalWechatIngress.collector.image`), + pullPolicy: stringField(image, "pullPolicy", `${configLabel}.personalWechatIngress.collector.image`), + context: stringField(image, "context", `${configLabel}.personalWechatIngress.collector.image`), + dockerfile: stringField(image, "dockerfile", `${configLabel}.personalWechatIngress.collector.image`), + wcferryVersion: stringField(image, "wcferryVersion", `${configLabel}.personalWechatIngress.collector.image`), + }, wcfHost: stringField(collector, "wcfHost", `${configLabel}.personalWechatIngress.collector`), commandPort: numberField(collector, "commandPort", `${configLabel}.personalWechatIngress.collector`), messagePort: numberField(collector, "messagePort", `${configLabel}.personalWechatIngress.collector`), @@ -634,6 +1468,7 @@ function parsePersonalWechatIngress(raw: Record): WechatArchive queueMode: stringField(collector, "queueMode", `${configLabel}.personalWechatIngress.collector`), outboxMode: stringField(collector, "outboxMode", `${configLabel}.personalWechatIngress.collector`), archiveCallbackRef: stringField(collector, "archiveCallbackRef", `${configLabel}.personalWechatIngress.collector`), + secretName: stringField(collector, "secretName", `${configLabel}.personalWechatIngress.collector`), readOnly: { enabled: booleanField(readOnly, "enabled", `${configLabel}.personalWechatIngress.collector.readOnly`), sendCapability: booleanField(readOnly, "sendCapability", `${configLabel}.personalWechatIngress.collector.readOnly`), @@ -669,6 +1504,12 @@ function validatePersonalWechatIngress(config: WechatArchiveConfig["personalWech if (config.collector.storage.kind !== "PersistentVolumeClaim") { throw new Error(`${configLabel}.personalWechatIngress.collector.storage.kind must be PersistentVolumeClaim`); } + for (const repoPath of [config.collector.image.context, config.collector.image.dockerfile]) { + if (!existsSync(resolveRepoPath(repoPath))) throw new Error(`${configLabel}.personalWechatIngress.collector.image path does not exist: ${repoPath}`); + } + if (!["Always", "IfNotPresent", "Never"].includes(config.collector.image.pullPolicy)) { + throw new Error(`${configLabel}.personalWechatIngress.collector.image.pullPolicy must be Always, IfNotPresent, or Never`); + } if (config.pcWechat.requiredVersion !== config.wcfHost.requiredWechatVersion) { throw new Error(`${configLabel}.personalWechatIngress pcWechat.requiredVersion and wcfHost.requiredWechatVersion must match`); } @@ -701,6 +1542,7 @@ function personalWechatIngressSummary(config: WechatArchiveConfig["personalWecha isolation: config.pcWechat.isolation, requiredVersion: config.pcWechat.requiredVersion, installerAsset: config.pcWechat.installerAsset, + installMode: config.pcWechat.installMode, installRoot: config.pcWechat.installRoot, dataRoot: config.pcWechat.dataRoot, autoUpdatePolicy: config.pcWechat.autoUpdatePolicy, @@ -724,6 +1566,15 @@ function personalWechatIngressSummary(config: WechatArchiveConfig["personalWecha createNamespace: config.collector.createNamespace, workload: config.collector.workload, storage: config.collector.storage, + image: { + reference: collectorImageRef(config), + repository: config.collector.image.repository, + tag: config.collector.image.tag, + pullPolicy: config.collector.image.pullPolicy, + context: config.collector.image.context, + dockerfile: config.collector.image.dockerfile, + wcferryVersion: config.collector.image.wcferryVersion, + }, wcfHost: config.collector.wcfHost, commandPort: config.collector.commandPort, messagePort: config.collector.messagePort, @@ -731,6 +1582,7 @@ function personalWechatIngressSummary(config: WechatArchiveConfig["personalWecha queueMode: config.collector.queueMode, outboxMode: config.collector.outboxMode, archiveCallbackRef: config.collector.archiveCallbackRef, + secretName: config.collector.secretName, readOnly: { enabled: config.collector.readOnly.enabled, sendCapability: config.collector.readOnly.sendCapability, diff --git a/scripts/src/platform-infra.ts b/scripts/src/platform-infra.ts index 6054aa30..dc3dfda4 100644 --- a/scripts/src/platform-infra.ts +++ b/scripts/src/platform-infra.ts @@ -251,6 +251,11 @@ export function platformInfraHelp(): unknown { "bun scripts/cli.ts platform-infra wechat-archive status [--target G14] [--full|--raw]", "bun scripts/cli.ts platform-infra wechat-archive validate [--target G14] [--full|--raw]", "bun scripts/cli.ts platform-infra wechat-archive pull --remote-path /UniDesk/WeChatArchive/...", + "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm", + "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm", + "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm", + "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm", + "bun scripts/cli.ts platform-infra wechat-archive collector-status --full", ], description: "Operate YAML-controlled platform-infra services such as Sub2API, LangBot, n8n and WeChat archive workflows. Public services use PK01 Caddy+FRP rather than Kubernetes Ingress, NodePort, or LoadBalancer.", target: { diff --git a/scripts/windows/personal-wechat/prepare.ps1 b/scripts/windows/personal-wechat/prepare.ps1 new file mode 100644 index 00000000..5c1accef --- /dev/null +++ b/scripts/windows/personal-wechat/prepare.ps1 @@ -0,0 +1,115 @@ +$ErrorActionPreference = "Stop" + +$Root = "C:\UniDesk\personal-wechat" +$WechatRoot = Join-Path $Root "wechat-3.9.12.51" +$WcfRoot = Join-Path $Root "wcf\v39.5.2" +$StateRoot = Join-Path $Root "wcf-state" +$DownloadRoot = Join-Path $Root "downloads" +$PrepareResult = Join-Path $StateRoot "prepare-result.json" +$Python = "C:\ProgramData\miniconda3\python.exe" +$PipIndex = "http://mirrors.aliyun.com/pypi/simple/" +$ReleaseBase = "https://github.com/lich0821/WeChatFerry/releases/download/v39.5.2" + +New-Item -ItemType Directory -Force $WechatRoot,$WcfRoot,$StateRoot,$DownloadRoot | Out-Null + +function Download-IfMissing { + param([string]$Url, [string]$Path) + if (Test-Path $Path) { return } + $tmp = "$Path.part" + Invoke-WebRequest -Uri $Url -OutFile $tmp -UseBasicParsing + Move-Item -Force $tmp $Path +} + +function Find-WeChatExe { + $candidates = @( + (Join-Path $WechatRoot "WeChat.exe"), + "C:\Program Files (x86)\Tencent\WeChat\WeChat.exe", + "C:\Program Files\Tencent\WeChat\WeChat.exe" + ) + foreach ($candidate in $candidates) { + if (Test-Path $candidate) { return $candidate } + } + return $null +} + +$startedAt = (Get-Date).ToUniversalTime().ToString("o") +$installer = Join-Path $DownloadRoot "WeChatSetup-3.9.12.51.exe" +Download-IfMissing "$ReleaseBase/WeChatSetup-3.9.12.51.exe" $installer +Download-IfMissing "$ReleaseBase/sdk.dll" (Join-Path $WcfRoot "sdk.dll") +Download-IfMissing "$ReleaseBase/spy.dll" (Join-Path $WcfRoot "spy.dll") +Download-IfMissing "$ReleaseBase/spy_debug.dll" (Join-Path $WcfRoot "spy_debug.dll") + +if (!(Test-Path $Python)) { + throw "Expected Python not found: $Python" +} + +$pipOut = Join-Path $StateRoot "prepare-pip.stdout.log" +$pipErr = Join-Path $StateRoot "prepare-pip.stderr.log" +& $Python -m pip install --trusted-host mirrors.aliyun.com --index-url $PipIndex "wcferry==39.5.2.0" > $pipOut 2> $pipErr +if ($LASTEXITCODE -ne 0) { throw "pip install wcferry failed" } + +Copy-Item -Force "$PSScriptRoot\wcf_host.py" (Join-Path $WcfRoot "wcf_host.py") + +if (-not (Get-NetFirewallRule -DisplayName "UniDesk Personal WeChat WCF 10086" -ErrorAction SilentlyContinue)) { + New-NetFirewallRule ` + -DisplayName "UniDesk Personal WeChat WCF 10086" ` + -Direction Inbound ` + -Action Allow ` + -Protocol TCP ` + -LocalPort 10086,10087 ` + -RemoteAddress 172.26.0.0/16,10.42.0.0/16,127.0.0.1 ` + -Profile Any ` + -ErrorAction SilentlyContinue | Out-Null +} + +$wechatExe = Find-WeChatExe +$installAttempted = $false +$installExitCode = $null +$installStdout = Join-Path $StateRoot "prepare-wechat-installer.stdout.log" +$installStderr = Join-Path $StateRoot "prepare-wechat-installer.stderr.log" +if (-not $wechatExe) { + $installAttempted = $true + $args = @("/S", "/D=$WechatRoot") + $proc = Start-Process -FilePath $installer -ArgumentList $args -Wait -PassThru -RedirectStandardOutput $installStdout -RedirectStandardError $installStderr -WindowStyle Hidden + $installExitCode = $proc.ExitCode + Start-Sleep -Seconds 5 + $wechatExe = Find-WeChatExe +} + +$probePy = Join-Path $StateRoot "prepare-probe.py" +@' +import importlib.metadata +import json + +payload = {"ok": True} +try: + payload["wcferryVersion"] = importlib.metadata.version("wcferry") +except Exception as exc: + payload = {"ok": False, "error": f"{type(exc).__name__}: {exc}"} +print(json.dumps(payload, ensure_ascii=False)) +'@ | Set-Content -Encoding utf8 $probePy +$pyProbeJson = & $Python $probePy +$pyProbe = $pyProbeJson | ConvertFrom-Json + +$summary = [ordered]@{ + ok = [bool]($pyProbe.ok -and $wechatExe) + startedAt = $startedAt + finishedAt = (Get-Date).ToUniversalTime().ToString("o") + root = $Root + wechatInstaller = $installer + wechatRoot = $WechatRoot + wechatExe = $wechatExe + installAttempted = $installAttempted + installExitCode = $installExitCode + installLogs = [ordered]@{ + stdout = $installStdout + stderr = $installStderr + } + wcfRoot = $WcfRoot + stateRoot = $StateRoot + python = $Python + wcferryVersion = $pyProbe.wcferryVersion + next = if ($wechatExe) { "Run start.ps1 and scan the WeChat login QR." } else { "WeChat silent install did not produce WeChat.exe; run the installer UI from the interactive Windows session, then re-run prepare.ps1." } +} +$summary | ConvertTo-Json -Depth 6 | Set-Content -Encoding utf8 $PrepareResult +$summary | ConvertTo-Json -Depth 6 diff --git a/scripts/windows/personal-wechat/start.ps1 b/scripts/windows/personal-wechat/start.ps1 new file mode 100644 index 00000000..7970b379 --- /dev/null +++ b/scripts/windows/personal-wechat/start.ps1 @@ -0,0 +1,30 @@ +$ErrorActionPreference = "Stop" + +$Root = "C:\UniDesk\personal-wechat" +$WcfRoot = Join-Path $Root "wcf\v39.5.2" +$StateRoot = Join-Path $Root "wcf-state" +$Python = "C:\ProgramData\miniconda3\python.exe" +$Script = Join-Path $WcfRoot "wcf_host.py" +$PidFile = Join-Path $StateRoot "wcf-host.pid" +$Stdout = Join-Path $StateRoot "wcf-host.stdout.log" +$Stderr = Join-Path $StateRoot "wcf-host.stderr.log" + +New-Item -ItemType Directory -Force $StateRoot | Out-Null + +if (Test-Path $PidFile) { + $oldPid = Get-Content $PidFile -ErrorAction SilentlyContinue + if ($oldPid) { + $existing = Get-Process -Id ([int]$oldPid) -ErrorAction SilentlyContinue + if ($existing) { + [pscustomobject]@{ ok = $true; alreadyRunning = $true; pid = $existing.Id; stdout = $Stdout; stderr = $Stderr } | ConvertTo-Json -Depth 4 + exit 0 + } + } +} + +$env:WCF_COMMAND_PORT = "10086" +$env:WCF_STATE_ROOT = $StateRoot +$proc = Start-Process -FilePath $Python -ArgumentList @($Script) -WorkingDirectory $WcfRoot -WindowStyle Normal -PassThru -RedirectStandardOutput $Stdout -RedirectStandardError $Stderr +$proc.Id | Set-Content -Encoding ascii $PidFile + +[pscustomobject]@{ ok = $true; started = $true; pid = $proc.Id; stdout = $Stdout; stderr = $Stderr; status = (Join-Path $StateRoot "status.json") } | ConvertTo-Json -Depth 4 diff --git a/scripts/windows/personal-wechat/status.ps1 b/scripts/windows/personal-wechat/status.ps1 new file mode 100644 index 00000000..4a51e746 --- /dev/null +++ b/scripts/windows/personal-wechat/status.ps1 @@ -0,0 +1,54 @@ +$ErrorActionPreference = "Continue" + +$Root = "C:\UniDesk\personal-wechat" +$StateRoot = Join-Path $Root "wcf-state" +$PidFile = Join-Path $StateRoot "wcf-host.pid" +$StatusFile = Join-Path $StateRoot "status.json" +$PrepareResultFile = Join-Path $StateRoot "prepare-result.json" +$Stdout = Join-Path $StateRoot "wcf-host.stdout.log" +$Stderr = Join-Path $StateRoot "wcf-host.stderr.log" +function Read-JsonFile { + param([string]$Path) + if (!(Test-Path $Path)) { return $null } + try { return Get-Content $Path -Raw | ConvertFrom-Json } catch { return @{ parseError = $_.Exception.Message } } +} +function Tail-Text { + param([string]$Path) + if (!(Test-Path $Path)) { return "" } + try { + return (Get-Content $Path -Tail 80 -ErrorAction SilentlyContinue) -join "`n" + } catch { + return $_.Exception.Message + } +} +$pidValue = $null +$running = $false +if (Test-Path $PidFile) { + $pidValue = Get-Content $PidFile -ErrorAction SilentlyContinue + if ($pidValue) { + $running = [bool](Get-Process -Id ([int]$pidValue) -ErrorAction SilentlyContinue) + } +} +$status = Read-JsonFile $StatusFile +$prepare = Read-JsonFile $PrepareResultFile +[pscustomobject]@{ + ok = $true + pid = $pidValue + running = $running + prepared = if ($prepare) { [bool]$prepare.ok } else { $false } + prepare = $prepare + status = $status + ports = Get-NetTCPConnection -LocalPort 10086,10087 -ErrorAction SilentlyContinue | Select-Object LocalAddress,LocalPort,State,OwningProcess + logs = [ordered]@{ + stdout = $Stdout + stderr = $Stderr + stdoutTail = Tail-Text $Stdout + stderrTail = Tail-Text $Stderr + } + paths = [ordered]@{ + root = $Root + stateRoot = $StateRoot + statusFile = $StatusFile + prepareResultFile = $PrepareResultFile + } +} | ConvertTo-Json -Depth 8 diff --git a/scripts/windows/personal-wechat/wcf_host.py b/scripts/windows/personal-wechat/wcf_host.py new file mode 100644 index 00000000..806f7d51 --- /dev/null +++ b/scripts/windows/personal-wechat/wcf_host.py @@ -0,0 +1,76 @@ +import json +import os +import signal +import sys +import time +from datetime import datetime, timezone + +from wcferry import Wcf + + +PORT = int(os.environ.get("WCF_COMMAND_PORT", "10086")) +STATE_ROOT = os.environ.get("WCF_STATE_ROOT", r"C:\UniDesk\personal-wechat\wcf-state") +DEBUG = os.environ.get("WCF_DEBUG", "false").lower() == "true" + +STOP = False + + +def now_iso(): + return datetime.now(timezone.utc).isoformat() + + +def log(event, **fields): + os.makedirs(STATE_ROOT, exist_ok=True) + payload = {"ts": now_iso(), "event": event, **fields} + line = json.dumps(payload, ensure_ascii=False, sort_keys=True) + print(line, flush=True) + with open(os.path.join(STATE_ROOT, "wcf-host.log"), "a", encoding="utf-8") as fp: + fp.write(line + "\n") + + +def handle_signal(_signum, _frame): + global STOP + STOP = True + + +signal.signal(signal.SIGTERM, handle_signal) +signal.signal(signal.SIGINT, handle_signal) + + +def main(): + log("wcf-host-start", port=PORT, debug=DEBUG) + wcf = Wcf(port=PORT, debug=DEBUG, block=False) + while not STOP: + try: + logged_in = bool(wcf.is_login()) + qrcode = "" + if not logged_in: + try: + qrcode = wcf.get_qrcode() + except Exception: + qrcode = "" + state = { + "ts": now_iso(), + "isLogin": logged_in, + "qrcodePresent": bool(qrcode), + "commandPort": PORT, + "messagePort": PORT + 1, + } + if qrcode: + state["qrcode"] = qrcode + with open(os.path.join(STATE_ROOT, "status.json"), "w", encoding="utf-8") as fp: + json.dump(state, fp, ensure_ascii=False, indent=2, sort_keys=True) + log("wcf-host-status", isLogin=logged_in, qrcodePresent=bool(qrcode)) + time.sleep(5) + except Exception as exc: + log("wcf-host-error", error=f"{type(exc).__name__}: {exc}"[:1000]) + time.sleep(3) + try: + wcf.cleanup() + except Exception: + pass + log("wcf-host-stop") + + +if __name__ == "__main__": + main() diff --git a/src/components/platform-infra/personal-wechat-collector/Dockerfile b/src/components/platform-infra/personal-wechat-collector/Dockerfile new file mode 100644 index 00000000..024c8317 --- /dev/null +++ b/src/components/platform-infra/personal-wechat-collector/Dockerfile @@ -0,0 +1,18 @@ +FROM python:3.12-alpine + +ARG WCFERRY_VERSION=39.5.2.0 + +RUN apk add --no-cache ca-certificates tzdata \ + && python -m pip install --no-cache-dir --trusted-host mirrors.aliyun.com \ + --index-url http://mirrors.aliyun.com/pypi/simple/ \ + "wcferry==${WCFERRY_VERSION}" \ + && python - <<'PY' +import importlib.metadata +print("wcferry", importlib.metadata.version("wcferry")) +PY + +WORKDIR /app +COPY collector.py /app/collector.py + +ENV PYTHONUNBUFFERED=1 +CMD ["python", "/app/collector.py"] diff --git a/src/components/platform-infra/personal-wechat-collector/collector.py b/src/components/platform-infra/personal-wechat-collector/collector.py new file mode 100644 index 00000000..745a369b --- /dev/null +++ b/src/components/platform-infra/personal-wechat-collector/collector.py @@ -0,0 +1,282 @@ +import hashlib +import json +import os +import queue +import signal +import sqlite3 +import sys +import time +import urllib.error +import urllib.request +from datetime import datetime, timezone +from zoneinfo import ZoneInfo + +from wcferry import Wcf + + +WCF_HOST = os.environ.get("WCF_HOST", "127.0.0.1") +WCF_COMMAND_PORT = int(os.environ.get("WCF_COMMAND_PORT", "10086")) +STATE_ROOT = os.environ.get("STATE_ROOT", "/var/lib/unidesk/personal-wechat") +ARCHIVE_CALLBACK_URL = os.environ.get("ARCHIVE_CALLBACK_URL", "") +ARCHIVE_CALLBACK_TOKEN = os.environ.get("ARCHIVE_CALLBACK_TOKEN", "") +ARCHIVE_REMOTE_ROOT = os.environ.get("ARCHIVE_REMOTE_ROOT", "/UniDesk/WeChatArchive") +ARCHIVE_TIMEZONE = os.environ.get("ARCHIVE_TIMEZONE", "Asia/Shanghai") +POLL_IDLE_SECONDS = float(os.environ.get("POLL_IDLE_SECONDS", "1")) +STATUS_INTERVAL_SECONDS = float(os.environ.get("STATUS_INTERVAL_SECONDS", "15")) + +STOP = False + + +def now_iso(): + return datetime.now(timezone.utc).isoformat() + + +def log(event, **fields): + payload = {"ts": now_iso(), "event": event, **fields} + print(json.dumps(payload, ensure_ascii=False, sort_keys=True), flush=True) + + +def handle_signal(signum, _frame): + global STOP + STOP = True + log("signal", signum=signum) + + +signal.signal(signal.SIGTERM, handle_signal) +signal.signal(signal.SIGINT, handle_signal) + + +def ensure_state(): + os.makedirs(STATE_ROOT, exist_ok=True) + os.makedirs(os.path.join(STATE_ROOT, "messages"), exist_ok=True) + db_path = os.path.join(STATE_ROOT, "collector.sqlite3") + conn = sqlite3.connect(db_path, timeout=30) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + observed_at TEXT NOT NULL, + sender TEXT, + roomid TEXT, + type INTEGER, + ts INTEGER, + content_sha256 TEXT, + payload_json TEXT NOT NULL, + callback_status INTEGER, + callback_error TEXT + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS collector_state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """ + ) + conn.commit() + return conn + + +def safe_message(msg): + content = str(getattr(msg, "content", "") or "") + xml = str(getattr(msg, "xml", "") or "") + extra = str(getattr(msg, "extra", "") or "") + thumb = str(getattr(msg, "thumb", "") or "") + summary = { + "id": str(getattr(msg, "id", "")), + "ts": int(getattr(msg, "ts", 0) or 0), + "type": int(getattr(msg, "type", 0) or 0), + "sender": str(getattr(msg, "sender", "") or ""), + "roomid": str(getattr(msg, "roomid", "") or ""), + "fromSelf": bool(msg.from_self()) if hasattr(msg, "from_self") else False, + "fromGroup": bool(msg.from_group()) if hasattr(msg, "from_group") else False, + "contentSha256": hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest(), + "contentPreview": content[:160], + "xmlSha256": hashlib.sha256(xml.encode("utf-8", errors="replace")).hexdigest() if xml else "", + "extraSha256": hashlib.sha256(extra.encode("utf-8", errors="replace")).hexdigest() if extra else "", + "thumbSha256": hashlib.sha256(thumb.encode("utf-8", errors="replace")).hexdigest() if thumb else "", + } + return summary, content + + +def sanitize_path_segment(value, fallback="unknown"): + text = str(value or fallback) + cleaned = [] + for char in text: + if char.isalnum() or char in "._-": + cleaned.append(char) + else: + cleaned.append("-") + normalized = "".join(cleaned).strip("-")[:120] + return normalized or fallback + + +def archive_date(summary): + try: + zone = ZoneInfo(ARCHIVE_TIMEZONE) + except Exception: + zone = timezone.utc + ts = int(summary.get("ts") or 0) + if ts > 0: + dt = datetime.fromtimestamp(ts, tz=zone) + else: + dt = datetime.now(zone) + return dt.strftime("%Y-%m-%d") + + +def archive_payload(summary, content): + message_id = summary["id"] or f"missing-{summary['ts']}-{summary['contentSha256'][:12]}" + conversation_id = summary["roomid"] or summary["sender"] or "unknown" + sender_id = summary["sender"] or "unknown" + message_type = "text" if summary["type"] == 1 else "wcf-message" + extension = "txt" + filename = f"{sanitize_path_segment(message_id, 'message')}.{extension}" + remote_path = "/".join([ + ARCHIVE_REMOTE_ROOT.rstrip("/"), + archive_date(summary), + sanitize_path_segment(conversation_id, "conversation"), + filename, + ]) + text = "\n".join([ + f"messageId={message_id}", + f"observedAt={now_iso()}", + f"sender={sender_id}", + f"roomid={summary['roomid']}", + f"type={summary['type']}", + f"timestamp={summary['ts']}", + f"contentSha256={summary['contentSha256']}", + "", + content, + ]) + return { + "messageId": message_id, + "conversationId": conversation_id, + "senderId": sender_id, + "messageType": message_type, + "text": text, + "message": { + "messageId": message_id, + "conversationId": conversation_id, + "senderId": sender_id, + "messageType": message_type, + "text": text, + "receivedAt": now_iso(), + }, + "archive": { + "remotePath": remote_path, + "filename": filename, + }, + "metadata": { + "source": "personal-wechat-collector", + "wcfType": summary["type"], + "contentSha256": summary["contentSha256"], + "fromSelf": summary["fromSelf"], + "fromGroup": summary["fromGroup"], + }, + } + + +def post_callback(payload): + if not ARCHIVE_CALLBACK_URL: + return None, "callback-url-empty" + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + req = urllib.request.Request(ARCHIVE_CALLBACK_URL, data=body, method="POST") + req.add_header("content-type", "application/json") + if ARCHIVE_CALLBACK_TOKEN: + req.add_header("authorization", f"Bearer {ARCHIVE_CALLBACK_TOKEN}") + req.add_header("x-unidesk-wechat-archive-token", ARCHIVE_CALLBACK_TOKEN) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + resp.read(1024) + return int(resp.status), "" + except urllib.error.HTTPError as exc: + return int(exc.code), str(exc)[:500] + except Exception as exc: + return None, f"{type(exc).__name__}: {exc}"[:500] + + +def save_message(conn, summary, callback_status, callback_error): + message_id = summary["id"] or f"missing-{summary['ts']}-{summary['contentSha256'][:12]}" + conn.execute( + """ + INSERT INTO messages ( + id, observed_at, sender, roomid, type, ts, content_sha256, payload_json, callback_status, callback_error + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + observed_at=excluded.observed_at, + callback_status=excluded.callback_status, + callback_error=excluded.callback_error + """, + ( + message_id, + now_iso(), + summary["sender"], + summary["roomid"], + summary["type"], + summary["ts"], + summary["contentSha256"], + json.dumps(summary, ensure_ascii=False, sort_keys=True), + callback_status, + callback_error, + ), + ) + conn.commit() + + +def set_state(conn, key, value): + conn.execute( + "INSERT INTO collector_state(key, value, updated_at) VALUES (?, ?, ?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at", + (key, str(value), now_iso()), + ) + conn.commit() + + +def main(): + conn = ensure_state() + log("collector-start", wcfHost=WCF_HOST, commandPort=WCF_COMMAND_PORT, stateRoot=STATE_ROOT, readOnly=True, sendCapability=False) + wcf = Wcf(host=WCF_HOST, port=WCF_COMMAND_PORT, debug=False, block=False) + receiving = False + last_status = 0.0 + while not STOP: + try: + logged_in = bool(wcf.is_login()) + set_state(conn, "is_login", "true" if logged_in else "false") + now = time.time() + if now - last_status >= STATUS_INTERVAL_SECONDS: + self_wxid = "" + if logged_in: + try: + self_wxid = wcf.get_self_wxid() + set_state(conn, "self_wxid_sha256", hashlib.sha256(self_wxid.encode()).hexdigest()) + except Exception as exc: + log("self-wxid-error", error=f"{type(exc).__name__}: {exc}"[:500]) + log("collector-status", isLogin=logged_in, receiving=receiving, selfWxidPresent=bool(self_wxid)) + last_status = now + if logged_in and not receiving: + receiving = bool(wcf.enable_receiving_msg()) + log("receiving-enabled", ok=receiving) + if receiving: + try: + msg = wcf.get_msg(block=True) + except queue.Empty: + continue + summary, content = safe_message(msg) + payload = archive_payload(summary, content) + status, error = post_callback(payload) + save_message(conn, summary, status, error) + log("message-archived", id=summary["id"], type=summary["type"], callbackStatus=status, callbackError=bool(error)) + else: + time.sleep(POLL_IDLE_SECONDS) + except Exception as exc: + log("collector-loop-error", error=f"{type(exc).__name__}: {exc}"[:1000]) + time.sleep(3) + log("collector-stop") + + +if __name__ == "__main__": + main()