feat: add wechat baidu archive workflow
This commit is contained in:
Codex
+7
-3
@@ -60,7 +60,7 @@ export function rootHelp(): unknown {
|
||||
{ command: "hwlab nodes control-plane|git-mirror|secret --node <node> --lane <lane>", description: "Manage HWLAB node/lane runtime prerequisites, including D601 YAML-declared infra/tools-image/Argo bootstrap and G14 v0.3+ runtime lanes, with the node identity passed as data." },
|
||||
{ command: "hwlab g14 monitor-prs | hwlab g14 control-plane status|apply|trigger-current|runtime-migration|cleanup-runs|cleanup-released-pvs | hwlab g14 git-mirror status|apply|sync|flush | hwlab g14 tools-image status|build", description: "Start the legacy G14 PR monitor, run bounded v0.2 Tekton/Argo control-plane, manual PipelineRun trigger, runtime migration, CI workspace retention, manual devops-infra git mirror/relay maintenance, or fixed HWLAB CI tools image actions; long confirmed trigger/sync/flush actions return async jobs by default." },
|
||||
{ command: "agentrun get|describe|events|logs|result|ack|cancel|dispatch|create|apply|send|control-plane|git-mirror", description: "Use AgentRun v0.1 resource primitives with low-noise human output by default; session follow-up uses send only and the server decides internal steer vs turn." },
|
||||
{ command: "platform-infra sub2api|langbot|n8n ...", description: "Deploy platform-infra services such as Sub2API, LangBot and n8n, manage YAML-controlled public FRP/Caddy exposure, and inspect status/logs without printing secrets." },
|
||||
{ command: "platform-infra sub2api|langbot|n8n|wechat-archive ...", description: "Deploy platform-infra services such as Sub2API, LangBot and n8n, manage YAML-controlled public FRP/Caddy exposure and WeChat archive workflows, and inspect status/logs without printing secrets." },
|
||||
{ command: "platform-db postgres plan|status|apply", description: "Manage YAML-declared host-native PostgreSQL 16 on PK01 for Sub2API/platform state, with PostgreSQL native TLS and redacted secret exports." },
|
||||
{ command: "hwlab cd audit --env dev | hwlab cd status --env dev | hwlab cd apply --env dev --dry-run", description: "Legacy D601 HWLAB DEV CD wrapper kept for explicit old-path diagnostics; current HWLAB rollout uses G14 GitOps." },
|
||||
{ command: "code-agent-sandbox", description: "Independent Code Agent Sandbox service skeleton for adapter, mode, and credential-boundary diagnostics." },
|
||||
@@ -605,7 +605,7 @@ function agentRunHelpSummary(): unknown {
|
||||
|
||||
function platformInfraHelpSummary(): unknown {
|
||||
return {
|
||||
command: "platform-infra sub2api|langbot|n8n ...",
|
||||
command: "platform-infra sub2api|langbot|n8n|wechat-archive ...",
|
||||
output: "json",
|
||||
usage: [
|
||||
"bun scripts/cli.ts platform-infra sub2api plan",
|
||||
@@ -623,8 +623,12 @@ function platformInfraHelpSummary(): unknown {
|
||||
"bun scripts/cli.ts platform-infra n8n apply --confirm",
|
||||
"bun scripts/cli.ts platform-infra n8n status",
|
||||
"bun scripts/cli.ts platform-infra n8n validate --full",
|
||||
"bun scripts/cli.ts platform-infra wechat-archive plan",
|
||||
"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/...",
|
||||
],
|
||||
description: "Operate G14 platform-infra services such as Sub2API, LangBot, n8n, and the YAML-controlled Codex pool.",
|
||||
description: "Operate G14 platform-infra services such as Sub2API, LangBot, n8n, WeChat archive workflows, and the YAML-controlled Codex pool.",
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,435 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
|
||||
import { basename, dirname, join, normalize, relative } from "node:path";
|
||||
import pathPosix from "node:path/posix";
|
||||
import type { UniDeskConfig } from "./config";
|
||||
import { rootPath } from "./config";
|
||||
import { coreInternalFetch } from "./microservices";
|
||||
import { capture, compactCapture, parseJsonOutput, redactText, shQuote } from "./platform-infra-public-service";
|
||||
|
||||
export interface OpsCommonOptions {
|
||||
targetId: string;
|
||||
full: boolean;
|
||||
raw: boolean;
|
||||
}
|
||||
|
||||
export interface OpsApplyOptions extends OpsCommonOptions {
|
||||
confirm: boolean;
|
||||
dryRun: boolean;
|
||||
wait: boolean;
|
||||
}
|
||||
|
||||
export interface OpsCommandOptionSpec {
|
||||
stringOptions?: string[];
|
||||
flagOptions?: string[];
|
||||
}
|
||||
|
||||
export function parseOpsCommonOptions(args: string[], spec: OpsCommandOptionSpec = {}): OpsCommonOptions & Record<string, string | boolean> {
|
||||
const stringOptions = new Set(["--target", ...(spec.stringOptions ?? [])]);
|
||||
const flagOptions = new Set(["--full", "--raw", ...(spec.flagOptions ?? [])]);
|
||||
const values: Record<string, string | boolean> = { targetId: "G14", full: false, raw: false };
|
||||
for (let index = 0; index < args.length; index += 1) {
|
||||
const arg = args[index];
|
||||
if (stringOptions.has(arg)) {
|
||||
const value = args[index + 1];
|
||||
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
|
||||
if (arg === "--target") values.targetId = value;
|
||||
else values[optionKey(arg)] = value;
|
||||
index += 1;
|
||||
} else if (flagOptions.has(arg)) {
|
||||
if (arg === "--full") values.full = true;
|
||||
else if (arg === "--raw") {
|
||||
values.raw = true;
|
||||
values.full = true;
|
||||
} else {
|
||||
values[optionKey(arg)] = true;
|
||||
}
|
||||
} else {
|
||||
throw new Error(`unsupported option: ${arg}`);
|
||||
}
|
||||
}
|
||||
const targetId = String(values.targetId);
|
||||
if (!/^[A-Za-z0-9._-]+$/u.test(targetId)) throw new Error("--target must be a simple target id");
|
||||
return values as OpsCommonOptions & Record<string, string | boolean>;
|
||||
}
|
||||
|
||||
export function parseOpsApplyOptions(args: string[]): OpsApplyOptions {
|
||||
const commonArgs: string[] = [];
|
||||
let confirm = false;
|
||||
let dryRun = false;
|
||||
let wait = false;
|
||||
for (let index = 0; index < args.length; index += 1) {
|
||||
const arg = args[index];
|
||||
if (arg === "--confirm") confirm = true;
|
||||
else if (arg === "--dry-run") dryRun = true;
|
||||
else if (arg === "--wait") wait = true;
|
||||
else {
|
||||
commonArgs.push(arg);
|
||||
if (arg === "--target") {
|
||||
commonArgs.push(args[index + 1] ?? "");
|
||||
index += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (confirm && dryRun) throw new Error("apply accepts only one of --confirm or --dry-run");
|
||||
return { ...parseOpsCommonOptions(commonArgs), confirm, dryRun: dryRun || !confirm, wait };
|
||||
}
|
||||
|
||||
export function readYamlRecord<T = Record<string, unknown>>(path: string, expectedKind?: string): T {
|
||||
const parsed = Bun.YAML.parse(readFileSync(path, "utf8")) as unknown;
|
||||
const record = asRecord(parsed, path);
|
||||
if (expectedKind !== undefined && record.kind !== expectedKind) throw new Error(`${repoRelative(path)}.kind must be ${expectedKind}`);
|
||||
return record as T;
|
||||
}
|
||||
|
||||
export function asRecord(value: unknown, path: string): Record<string, unknown> {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${path} must be an object`);
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
export function recordField(obj: Record<string, unknown>, key: string, path: string): Record<string, unknown> {
|
||||
return asRecord(obj[key], `${path}.${key}`);
|
||||
}
|
||||
|
||||
export function stringField(obj: Record<string, unknown>, key: string, path: string): string {
|
||||
const value = obj[key];
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new Error(`${path}.${key} must be a non-empty string`);
|
||||
return value;
|
||||
}
|
||||
|
||||
export function optionalStringField(obj: Record<string, unknown>, key: string, path: string): string | undefined {
|
||||
const value = obj[key];
|
||||
if (value === undefined || value === null) return undefined;
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new Error(`${path}.${key} must be a non-empty string when set`);
|
||||
return value;
|
||||
}
|
||||
|
||||
export function booleanField(obj: Record<string, unknown>, key: string, path: string): boolean {
|
||||
const value = obj[key];
|
||||
if (typeof value !== "boolean") throw new Error(`${path}.${key} must be a boolean`);
|
||||
return value;
|
||||
}
|
||||
|
||||
export function numberField(obj: Record<string, unknown>, key: string, path: string): number {
|
||||
const value = obj[key];
|
||||
if (typeof value !== "number" || !Number.isFinite(value)) throw new Error(`${path}.${key} must be a finite number`);
|
||||
return value;
|
||||
}
|
||||
|
||||
export function arrayField(obj: Record<string, unknown>, key: string, path: string): Record<string, unknown>[] {
|
||||
const value = obj[key];
|
||||
if (!Array.isArray(value)) throw new Error(`${path}.${key} must be an array`);
|
||||
return value.map((item, index) => asRecord(item, `${path}.${key}[${index}]`));
|
||||
}
|
||||
|
||||
export function stringListField(obj: Record<string, unknown>, key: string, path: string): string[] {
|
||||
const value = obj[key];
|
||||
if (!Array.isArray(value) || value.some((item) => typeof item !== "string" || item.length === 0)) throw new Error(`${path}.${key} must be an array of non-empty strings`);
|
||||
return value as string[];
|
||||
}
|
||||
|
||||
export function repoRelative(path: string): string {
|
||||
const rel = relative(rootPath(), path);
|
||||
return rel.startsWith("..") ? path : rel;
|
||||
}
|
||||
|
||||
export function resolveRepoPath(path: string): string {
|
||||
if (path.startsWith("/")) throw new Error(`repo-owned path must be relative: ${path}`);
|
||||
if (path.includes("..")) throw new Error(`repo-owned path must not contain ..: ${path}`);
|
||||
return rootPath(path);
|
||||
}
|
||||
|
||||
export function sanitizePathSegment(value: string, fallback = "unknown"): string {
|
||||
const cleaned = value.trim().replace(/[^A-Za-z0-9._-]+/gu, "-").replace(/^-+|-+$/gu, "");
|
||||
return cleaned.length > 0 ? cleaned.slice(0, 120) : fallback;
|
||||
}
|
||||
|
||||
export function dateInTimeZone(date: Date, timeZone: string): string {
|
||||
const parts = new Intl.DateTimeFormat("en-CA", {
|
||||
timeZone,
|
||||
year: "numeric",
|
||||
month: "2-digit",
|
||||
day: "2-digit",
|
||||
}).formatToParts(date);
|
||||
const pick = (type: string) => parts.find((part) => part.type === type)?.value ?? "00";
|
||||
return `${pick("year")}-${pick("month")}-${pick("day")}`;
|
||||
}
|
||||
|
||||
export function sha256File(path: string): string {
|
||||
return createHash("sha256").update(readFileSync(path)).digest("hex");
|
||||
}
|
||||
|
||||
export function ensureFileInside(root: string, relativePath: string): string {
|
||||
if (relativePath.startsWith("/") || relativePath.includes("..")) throw new Error(`path must be relative inside staging: ${relativePath}`);
|
||||
const rootAbsolute = hostRootPath(root);
|
||||
const resolved = normalize(join(rootAbsolute, relativePath));
|
||||
if (!resolved.startsWith(normalize(rootAbsolute + "/")) && resolved !== normalize(rootAbsolute)) throw new Error(`path escapes staging root: ${relativePath}`);
|
||||
mkdirSync(dirname(resolved), { recursive: true });
|
||||
return resolved;
|
||||
}
|
||||
|
||||
export function writeStagingText(params: { hostRoot: string; relativePath: string; content: string }): { hostPath: string; bytes: number; sha256: string } {
|
||||
const hostPath = ensureFileInside(params.hostRoot, params.relativePath);
|
||||
writeFileSync(hostPath, params.content, "utf8");
|
||||
return { hostPath, bytes: Buffer.byteLength(params.content, "utf8"), sha256: sha256File(hostPath) };
|
||||
}
|
||||
|
||||
export function writeStagingBase64(params: { hostRoot: string; relativePath: string; dataBase64: string }): { hostPath: string; bytes: number; sha256: string } {
|
||||
const buffer = Buffer.from(params.dataBase64, "base64");
|
||||
const hostPath = ensureFileInside(params.hostRoot, params.relativePath);
|
||||
writeFileSync(hostPath, buffer);
|
||||
return { hostPath, bytes: buffer.byteLength, sha256: sha256File(hostPath) };
|
||||
}
|
||||
|
||||
export function containerPathToHostPath(containerRoot: string, hostRoot: string, containerPath: string): string | null {
|
||||
const cleanContainerRoot = normalize(containerRoot);
|
||||
const cleanPath = normalize(containerPath);
|
||||
if (cleanPath !== cleanContainerRoot && !cleanPath.startsWith(`${cleanContainerRoot}/`)) return null;
|
||||
const rel = relative(cleanContainerRoot, cleanPath);
|
||||
return join(hostRootPath(hostRoot), rel);
|
||||
}
|
||||
|
||||
export function hostRootPath(root: string): string {
|
||||
return normalize(root.startsWith("/") ? root : rootPath(root));
|
||||
}
|
||||
|
||||
export interface MicroserviceProxyResponse {
|
||||
ok: boolean;
|
||||
status: number | null;
|
||||
body: unknown;
|
||||
raw: Record<string, unknown> | null;
|
||||
}
|
||||
|
||||
export function microserviceProxy(serviceId: string, upstreamPath: string, init: { method?: string; body?: unknown; timeoutMs?: number; maxResponseBytes?: number } = {}): MicroserviceProxyResponse {
|
||||
if (!upstreamPath.startsWith("/")) throw new Error("microservice upstream path must start with /");
|
||||
const response = coreInternalFetch(`/api/microservices/${encodeURIComponent(serviceId)}/proxy${upstreamPath}`, {
|
||||
method: init.method,
|
||||
body: init.body,
|
||||
timeoutMs: init.timeoutMs,
|
||||
maxResponseBytes: init.maxResponseBytes ?? 5_000_000,
|
||||
});
|
||||
const raw = typeof response === "object" && response !== null && !Array.isArray(response) ? response as Record<string, unknown> : null;
|
||||
return {
|
||||
ok: raw?.ok === true,
|
||||
status: typeof raw?.status === "number" ? raw.status : null,
|
||||
body: raw !== null && "body" in raw ? raw.body : response,
|
||||
raw,
|
||||
};
|
||||
}
|
||||
|
||||
export function assertProxyOk(response: MicroserviceProxyResponse, context: string): Record<string, unknown> {
|
||||
if (response.ok && typeof response.body === "object" && response.body !== null && !Array.isArray(response.body)) return response.body as Record<string, unknown>;
|
||||
throw new Error(`${context} failed: ${JSON.stringify(compactProxyResponse(response)).slice(0, 1200)}`);
|
||||
}
|
||||
|
||||
export function compactProxyResponse(response: MicroserviceProxyResponse): Record<string, unknown> {
|
||||
return {
|
||||
ok: response.ok,
|
||||
status: response.status,
|
||||
body: compactUnknown(response.body),
|
||||
};
|
||||
}
|
||||
|
||||
export function compactUnknown(value: unknown, maxString = 400): unknown {
|
||||
if (typeof value === "string") return redactText(value).slice(0, maxString);
|
||||
if (Array.isArray(value)) return { arrayPreview: value.slice(0, 5).map((item) => compactUnknown(item, maxString)), count: value.length };
|
||||
if (typeof value !== "object" || value === null) return value;
|
||||
const record = value as Record<string, unknown>;
|
||||
const output: Record<string, unknown> = {};
|
||||
for (const key of Object.keys(record).slice(0, 20)) output[key] = redactSensitiveKey(key) ? "<redacted>" : compactUnknown(record[key], maxString);
|
||||
if (Object.keys(record).length > 20) output.omittedKeys = Object.keys(record).length - 20;
|
||||
return output;
|
||||
}
|
||||
|
||||
export function redactSensitiveUnknown(value: unknown): unknown {
|
||||
if (typeof value === "string") return redactText(value);
|
||||
if (Array.isArray(value)) return value.map((item) => redactSensitiveUnknown(item));
|
||||
if (typeof value !== "object" || value === null) return value;
|
||||
const output: Record<string, unknown> = {};
|
||||
for (const [key, nested] of Object.entries(value as Record<string, unknown>)) output[key] = redactSensitiveKey(key) ? "<redacted>" : redactSensitiveUnknown(nested);
|
||||
return output;
|
||||
}
|
||||
|
||||
export async function waitForBaiduTransfer(serviceId: string, jobId: string, options: { timeoutMs: number; pollIntervalMs: number }): Promise<Record<string, unknown>> {
|
||||
const deadline = Date.now() + options.timeoutMs;
|
||||
let last: Record<string, unknown> | null = null;
|
||||
while (Date.now() <= deadline) {
|
||||
const detail = assertProxyOk(microserviceProxy(serviceId, `/api/transfers/${encodeURIComponent(jobId)}`, { timeoutMs: 30_000 }), `baidu transfer ${jobId}`);
|
||||
const job = asRecord(detail.job, `transfer ${jobId}.job`);
|
||||
last = job;
|
||||
const status = String(job.status || "");
|
||||
if (status === "succeeded") return { ok: true, job, events: detail.events };
|
||||
if (status === "failed" || status === "canceled") return { ok: false, job, events: detail.events };
|
||||
await new Promise((resolve) => setTimeout(resolve, options.pollIntervalMs));
|
||||
}
|
||||
return { ok: false, timeout: true, job: last };
|
||||
}
|
||||
|
||||
export function findBaiduFileByRemotePath(serviceId: string, remotePath: string): Record<string, unknown> | null {
|
||||
const dir = pathPosix.dirname(remotePath);
|
||||
const name = pathPosix.basename(remotePath);
|
||||
const response = assertProxyOk(microserviceProxy(serviceId, `/api/files?dir=${encodeURIComponent(dir)}&limit=500&order=time&desc=1`, { timeoutMs: 60_000 }), "baidu list files");
|
||||
const files = Array.isArray(response.files) ? response.files : [];
|
||||
for (const item of files) {
|
||||
if (typeof item !== "object" || item === null || Array.isArray(item)) continue;
|
||||
const record = item as Record<string, unknown>;
|
||||
if (String(record.path || "") === remotePath || String(record.serverFilename || record.filename || record.name || "") === name) return record;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export interface N8nWorkflowSyncResult {
|
||||
ok: boolean;
|
||||
mode: "dry-run" | "confirmed";
|
||||
workflow: { id: string; name: string; webhookPath: string; active: boolean };
|
||||
remote?: Record<string, unknown>;
|
||||
manifest?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export async function syncN8nWorkflow(config: UniDeskConfig, params: {
|
||||
targetRoute: string;
|
||||
namespace: string;
|
||||
deploymentName: string;
|
||||
workflowJson: Record<string, unknown>;
|
||||
workflowId: string;
|
||||
workflowName: string;
|
||||
webhookPath: string;
|
||||
active: boolean;
|
||||
dryRun: boolean;
|
||||
}): Promise<N8nWorkflowSyncResult> {
|
||||
if (params.dryRun) {
|
||||
return {
|
||||
ok: true,
|
||||
mode: "dry-run",
|
||||
workflow: { id: params.workflowId, name: params.workflowName, webhookPath: params.webhookPath, active: params.active },
|
||||
manifest: {
|
||||
nodes: Array.isArray(params.workflowJson.nodes) ? params.workflowJson.nodes.length : 0,
|
||||
active: params.workflowJson.active,
|
||||
settings: params.workflowJson.settings,
|
||||
},
|
||||
};
|
||||
}
|
||||
const encoded = Buffer.from(JSON.stringify(params.workflowJson, null, 2), "utf8").toString("base64");
|
||||
const script = `
|
||||
set -u
|
||||
tmp="$(mktemp -d)"
|
||||
trap 'rm -rf "$tmp"' EXIT
|
||||
workflow="$tmp/wechat-archive-workflow.json"
|
||||
printf '%s' '${encoded}' | base64 -d >"$workflow"
|
||||
kubectl -n ${params.namespace} exec -i deploy/${params.deploymentName} -c n8n -- sh -lc ${shQuote(`
|
||||
set -u
|
||||
mkdir -p /tmp/unidesk-n8n
|
||||
cat >/tmp/unidesk-n8n/wechat-archive-workflow.json
|
||||
n8n import:workflow --input=/tmp/unidesk-n8n/wechat-archive-workflow.json
|
||||
`)} <"$workflow" >"$tmp/import.out" 2>"$tmp/import.err"
|
||||
import_rc=$?
|
||||
if [ "$import_rc" -eq 0 ] && [ "${params.active ? "1" : "0"}" = "1" ]; then
|
||||
kubectl -n ${params.namespace} exec deploy/${params.deploymentName} -c n8n -- n8n update:workflow --id ${shQuote(params.workflowId)} --active=true >"$tmp/activate.out" 2>"$tmp/activate.err"
|
||||
activate_rc=$?
|
||||
else
|
||||
: >"$tmp/activate.out"
|
||||
if [ "$import_rc" -eq 0 ]; then
|
||||
printf '%s\\n' 'workflow activation not requested by YAML' >"$tmp/activate.err"
|
||||
activate_rc=0
|
||||
else
|
||||
printf '%s\\n' 'skipped because import failed' >"$tmp/activate.err"
|
||||
activate_rc=1
|
||||
fi
|
||||
fi
|
||||
if [ "$activate_rc" -eq 0 ] && [ "${params.active ? "1" : "0"}" = "1" ]; then
|
||||
kubectl -n ${params.namespace} rollout restart deployment/${params.deploymentName} >"$tmp/restart.out" 2>"$tmp/restart.err"
|
||||
restart_rc=$?
|
||||
if [ "$restart_rc" -eq 0 ]; then
|
||||
kubectl -n ${params.namespace} rollout status deployment/${params.deploymentName} --timeout=120s >"$tmp/rollout.out" 2>"$tmp/rollout.err"
|
||||
rollout_rc=$?
|
||||
else
|
||||
: >"$tmp/rollout.out"
|
||||
printf '%s\\n' 'skipped because rollout restart failed' >"$tmp/rollout.err"
|
||||
rollout_rc=1
|
||||
fi
|
||||
else
|
||||
: >"$tmp/restart.out"
|
||||
: >"$tmp/rollout.out"
|
||||
printf '%s\\n' 'skipped because activation failed or workflow is inactive by YAML' >"$tmp/restart.err"
|
||||
printf '%s\\n' 'skipped because activation failed or workflow is inactive by YAML' >"$tmp/rollout.err"
|
||||
restart_rc=0
|
||||
rollout_rc=0
|
||||
fi
|
||||
kubectl -n ${params.namespace} exec deploy/${params.deploymentName} -c n8n -- n8n list:workflow >"$tmp/list.out" 2>"$tmp/list.err"
|
||||
list_rc=$?
|
||||
python3 - "$import_rc" "$activate_rc" "$restart_rc" "$rollout_rc" "$list_rc" "$tmp/import.out" "$tmp/import.err" "$tmp/activate.out" "$tmp/activate.err" "$tmp/restart.out" "$tmp/restart.err" "$tmp/rollout.out" "$tmp/rollout.err" "$tmp/list.out" "$tmp/list.err" <<'PY'
|
||||
import json, sys
|
||||
def text(path, limit=6000):
|
||||
try:
|
||||
return open(path, encoding="utf-8", errors="replace").read()[-limit:]
|
||||
except FileNotFoundError:
|
||||
return ""
|
||||
payload = {
|
||||
"ok": int(sys.argv[1]) == 0 and int(sys.argv[2]) == 0 and int(sys.argv[3]) == 0 and int(sys.argv[4]) == 0,
|
||||
"workflow": {"id": "${params.workflowId}", "name": "${params.workflowName}", "webhookPath": "${params.webhookPath}", "active": ${params.active ? "True" : "False"}},
|
||||
"steps": {
|
||||
"import": {"exitCode": int(sys.argv[1]), "stdout": text(sys.argv[6]), "stderr": text(sys.argv[7])},
|
||||
"activate": {"exitCode": int(sys.argv[2]), "stdout": text(sys.argv[8]), "stderr": text(sys.argv[9])},
|
||||
"restart": {"exitCode": int(sys.argv[3]), "stdout": text(sys.argv[10]), "stderr": text(sys.argv[11])},
|
||||
"rollout": {"exitCode": int(sys.argv[4]), "stdout": text(sys.argv[12]), "stderr": text(sys.argv[13])},
|
||||
"list": {"exitCode": int(sys.argv[5]), "stdout": text(sys.argv[14]), "stderr": text(sys.argv[15])},
|
||||
},
|
||||
}
|
||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
sys.exit(0 if payload["ok"] else 1)
|
||||
PY
|
||||
`;
|
||||
const result = await capture(config, params.targetRoute, ["script"], script);
|
||||
const parsed = parseJsonOutput(result.stdout);
|
||||
return {
|
||||
ok: result.exitCode === 0 && parsed?.ok === true,
|
||||
mode: "confirmed",
|
||||
workflow: { id: params.workflowId, name: params.workflowName, webhookPath: params.webhookPath, active: params.active },
|
||||
remote: parsed ?? compactCapture(result, { full: true }),
|
||||
};
|
||||
}
|
||||
|
||||
export async function fetchJsonWithTimeout(url: string, body: unknown, timeoutMs: number): Promise<Record<string, unknown>> {
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
const response = await fetch(url, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
signal: controller.signal,
|
||||
});
|
||||
const text = await response.text();
|
||||
let parsed: unknown = text;
|
||||
try {
|
||||
parsed = text.length > 0 ? JSON.parse(text) as unknown : null;
|
||||
} catch {
|
||||
parsed = text;
|
||||
}
|
||||
return { ok: response.ok, status: response.status, body: parsed };
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
export function renderTemplate(template: string, values: Record<string, string>): string {
|
||||
return template.replace(/\{\{\s*([A-Za-z0-9_]+)\s*\}\}/gu, (_match, key: string) => values[key] ?? "");
|
||||
}
|
||||
|
||||
export function normalizeRemotePath(path: string): string {
|
||||
const normalized = pathPosix.normalize(path);
|
||||
if (!normalized.startsWith("/")) return `/${normalized}`;
|
||||
return normalized;
|
||||
}
|
||||
|
||||
export function relativeStagingPath(dir: string, filename: string): string {
|
||||
return pathPosix.join(dir.replace(/^\/+|\/+$/gu, ""), basename(filename));
|
||||
}
|
||||
|
||||
function optionKey(arg: string): string {
|
||||
return arg.replace(/^--/u, "").replace(/-([a-z])/gu, (_match, letter: string) => letter.toUpperCase());
|
||||
}
|
||||
|
||||
function redactSensitiveKey(key: string): boolean {
|
||||
return /(?:password|secret|token|api[_-]?key|authorization|cookie|qrcode|usercode|verificationurl|dlink|thumbs)/iu.test(key);
|
||||
}
|
||||
@@ -0,0 +1,729 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { existsSync } from "node:fs";
|
||||
import pathPosix from "node:path/posix";
|
||||
import type { UniDeskConfig } from "./config";
|
||||
import { rootPath } from "./config";
|
||||
import { startJob } from "./jobs";
|
||||
import {
|
||||
asRecord,
|
||||
assertProxyOk,
|
||||
booleanField,
|
||||
compactProxyResponse,
|
||||
compactUnknown,
|
||||
containerPathToHostPath,
|
||||
dateInTimeZone,
|
||||
fetchJsonWithTimeout,
|
||||
findBaiduFileByRemotePath,
|
||||
hostRootPath,
|
||||
microserviceProxy,
|
||||
normalizeRemotePath,
|
||||
numberField,
|
||||
optionalStringField,
|
||||
parseOpsApplyOptions,
|
||||
parseOpsCommonOptions,
|
||||
readYamlRecord,
|
||||
redactSensitiveUnknown,
|
||||
recordField,
|
||||
relativeStagingPath,
|
||||
renderTemplate,
|
||||
repoRelative,
|
||||
resolveRepoPath,
|
||||
sanitizePathSegment,
|
||||
sha256File,
|
||||
stringField,
|
||||
syncN8nWorkflow,
|
||||
waitForBaiduTransfer,
|
||||
writeStagingBase64,
|
||||
writeStagingText,
|
||||
type OpsApplyOptions,
|
||||
type OpsCommonOptions,
|
||||
} from "./platform-infra-ops-library";
|
||||
|
||||
const configFile = rootPath("config", "platform-infra", "wechat-archive.yaml");
|
||||
const configLabel = "config/platform-infra/wechat-archive.yaml";
|
||||
|
||||
interface WechatArchiveConfig {
|
||||
version: number;
|
||||
kind: "platform-infra-wechat-archive";
|
||||
metadata: { id: string; owner: string; relatedIssues: number[] };
|
||||
target: { id: string; route: string; namespace: string };
|
||||
langbot: { configRef: string; publicBaseUrl: string; expectedAdapter: string; callbackPath: string; notes: string[] };
|
||||
n8n: {
|
||||
configRef: string;
|
||||
publicBaseUrl: string;
|
||||
workflow: { name: string; id: string; webhookPath: string; active: boolean; timezone: string };
|
||||
};
|
||||
baiduNetdisk: {
|
||||
serviceId: string;
|
||||
proxyMode: string;
|
||||
configRef: string;
|
||||
staging: { hostRoot: string; containerRoot: string; outboxDir: string; pullDir: string };
|
||||
archive: {
|
||||
remoteRoot: string;
|
||||
timezone: string;
|
||||
textExtension: string;
|
||||
imageExtension: string;
|
||||
pathTemplate: string;
|
||||
maxInlineBytes: number;
|
||||
};
|
||||
};
|
||||
messageTypes: Record<string, { enabled: boolean; contentField: string; mediaMode: string; filenameField?: string; mimeTypeField?: string }>;
|
||||
validate: {
|
||||
timeoutMs: number;
|
||||
pollIntervalMs: number;
|
||||
fixtures: {
|
||||
text: { conversationId: string; senderId: string; text: string };
|
||||
image: { conversationId: string; senderId: string; filename: string; mimeType: string; dataBase64: string };
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
interface PullOptions extends OpsCommonOptions {
|
||||
remotePath?: string;
|
||||
fsId?: string;
|
||||
localPath?: string;
|
||||
}
|
||||
|
||||
export function wechatArchiveHelp(): Record<string, unknown> {
|
||||
return {
|
||||
command: "platform-infra wechat-archive plan|apply|status|validate|pull",
|
||||
output: "json",
|
||||
usage: [
|
||||
"bun scripts/cli.ts platform-infra wechat-archive plan [--target G14]",
|
||||
"bun scripts/cli.ts platform-infra wechat-archive apply [--target G14] --dry-run",
|
||||
"bun scripts/cli.ts platform-infra wechat-archive apply [--target G14] --confirm",
|
||||
"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/... [--target G14] [--full|--raw]",
|
||||
"bun scripts/cli.ts platform-infra wechat-archive pull --fs-id <baiduFsId> [--local-path wechat-archive/pulls/file] [--target G14]",
|
||||
],
|
||||
configTruth: configLabel,
|
||||
boundary: {
|
||||
langbot: "Receives WeChat/OpenClaw events and forwards archive events to the YAML-declared n8n webhook.",
|
||||
n8n: "Stores a YAML-rendered workflow that normalizes WeChat payloads.",
|
||||
baiduNetdisk: "Uploads and downloads through the private backend-core microservice proxy; Baidu credentials remain in the baidu-netdisk service.",
|
||||
database: "LangBot and n8n continue to use the single PK01/Pika01 external PostgreSQL instance with separate databases/roles.",
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export async function runWechatArchiveCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown>> {
|
||||
const [action = "plan"] = args;
|
||||
if (action === "plan") return plan(parseOpsCommonOptions(args.slice(1)));
|
||||
if (action === "apply") return await apply(config, parseOpsApplyOptions(args.slice(1)));
|
||||
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)));
|
||||
return { ok: false, error: "unsupported-platform-infra-wechat-archive-command", args, help: wechatArchiveHelp() };
|
||||
}
|
||||
|
||||
function parsePullOptions(args: string[]): PullOptions {
|
||||
const parsed = parseOpsCommonOptions(args, { stringOptions: ["--remote-path", "--fs-id", "--local-path"] });
|
||||
return {
|
||||
targetId: parsed.targetId,
|
||||
full: parsed.full,
|
||||
raw: parsed.raw,
|
||||
remotePath: typeof parsed.remotePath === "string" ? parsed.remotePath : undefined,
|
||||
fsId: typeof parsed.fsId === "string" ? parsed.fsId : undefined,
|
||||
localPath: typeof parsed.localPath === "string" ? parsed.localPath : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
function plan(options: OpsCommonOptions): Record<string, unknown> {
|
||||
const archive = readConfig();
|
||||
assertTarget(archive, options.targetId);
|
||||
const workflow = renderN8nWorkflow(archive);
|
||||
const policy = policyChecks(archive);
|
||||
return {
|
||||
ok: policy.every((check) => check.ok),
|
||||
action: "platform-infra-wechat-archive-plan",
|
||||
mutation: false,
|
||||
config: configSummary(archive),
|
||||
workflow: {
|
||||
id: archive.n8n.workflow.id,
|
||||
name: archive.n8n.workflow.name,
|
||||
webhookUrl: webhookUrl(archive),
|
||||
nodes: Array.isArray(workflow.nodes) ? workflow.nodes.length : 0,
|
||||
active: archive.n8n.workflow.active,
|
||||
},
|
||||
policy,
|
||||
next: {
|
||||
dryRun: `bun scripts/cli.ts platform-infra wechat-archive apply --target ${archive.target.id} --dry-run`,
|
||||
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}`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function apply(config: UniDeskConfig, options: OpsApplyOptions): Promise<Record<string, unknown>> {
|
||||
const archive = readConfig();
|
||||
assertTarget(archive, options.targetId);
|
||||
const policy = policyChecks(archive);
|
||||
if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-wechat-archive-apply", mode: "policy-blocked", policy };
|
||||
if (options.confirm && !options.wait) {
|
||||
const job = startJob(
|
||||
"platform_infra_wechat_archive_apply_g14",
|
||||
["bun", "scripts/cli.ts", "platform-infra", "wechat-archive", "apply", "--target", archive.target.id, "--confirm", "--wait"],
|
||||
"Sync the YAML-rendered n8n workflow for WeChat -> Baidu Netdisk archive",
|
||||
);
|
||||
return {
|
||||
ok: true,
|
||||
action: "platform-infra-wechat-archive-apply",
|
||||
mode: "async-job",
|
||||
mutation: true,
|
||||
job,
|
||||
statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`,
|
||||
};
|
||||
}
|
||||
const workflowJson = renderN8nWorkflow(archive);
|
||||
const sync = await syncN8nWorkflow(config, {
|
||||
targetRoute: archive.target.route,
|
||||
namespace: archive.target.namespace,
|
||||
deploymentName: "n8n",
|
||||
workflowJson,
|
||||
workflowId: archive.n8n.workflow.id,
|
||||
workflowName: archive.n8n.workflow.name,
|
||||
webhookPath: archive.n8n.workflow.webhookPath,
|
||||
active: archive.n8n.workflow.active,
|
||||
dryRun: options.dryRun,
|
||||
});
|
||||
return {
|
||||
ok: sync.ok,
|
||||
action: "platform-infra-wechat-archive-apply",
|
||||
mode: options.dryRun ? "dry-run" : "confirmed",
|
||||
mutation: !options.dryRun,
|
||||
config: configSummary(archive),
|
||||
policy,
|
||||
n8nWorkflow: sync,
|
||||
next: {
|
||||
status: `bun scripts/cli.ts platform-infra wechat-archive status --target ${archive.target.id}`,
|
||||
validate: `bun scripts/cli.ts platform-infra wechat-archive validate --target ${archive.target.id} --full`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function status(options: OpsCommonOptions): Promise<Record<string, unknown>> {
|
||||
const archive = readConfig();
|
||||
assertTarget(archive, options.targetId);
|
||||
const health = microserviceProxy(archive.baiduNetdisk.serviceId, "/health", { timeoutMs: 30_000 });
|
||||
const auth = microserviceProxy(archive.baiduNetdisk.serviceId, "/api/auth/status", { timeoutMs: 30_000 });
|
||||
const transfers = microserviceProxy(archive.baiduNetdisk.serviceId, "/api/transfers?limit=10", { timeoutMs: 30_000 });
|
||||
return {
|
||||
ok: health.ok && auth.ok,
|
||||
action: "platform-infra-wechat-archive-status",
|
||||
config: configSummary(archive),
|
||||
n8n: {
|
||||
webhookUrl: webhookUrl(archive),
|
||||
workflowId: archive.n8n.workflow.id,
|
||||
activeDesired: archive.n8n.workflow.active,
|
||||
},
|
||||
baiduNetdisk: {
|
||||
health: options.full ? redactSensitiveUnknown(health.raw) : compactProxyResponse(health),
|
||||
auth: options.full ? redactSensitiveUnknown(auth.raw) : compactProxyResponse(auth),
|
||||
recentTransfers: options.full ? redactSensitiveUnknown(transfers.raw) : compactProxyResponse(transfers),
|
||||
},
|
||||
valuesPrinted: false,
|
||||
...(options.raw ? { raw: redactSensitiveUnknown({ health, auth, transfers }) } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
async function validate(options: OpsCommonOptions): Promise<Record<string, unknown>> {
|
||||
const archive = readConfig();
|
||||
assertTarget(archive, options.targetId);
|
||||
const observedAt = new Date();
|
||||
const textPayload = fixturePayload(archive, "text", observedAt);
|
||||
const imagePayload = fixturePayload(archive, "image", observedAt);
|
||||
const n8nText = await callWorkflow(archive, textPayload);
|
||||
const n8nImage = await callWorkflow(archive, imagePayload);
|
||||
const textArchive = await archiveMessage(archive, textPayload, n8nText);
|
||||
const imageArchive = await archiveMessage(archive, imagePayload, n8nImage);
|
||||
const textPull = await pullByFsId(archive, String(textArchive.fsId || ""), pullLocalRel(archive, textArchive.remotePath));
|
||||
const imagePull = await pullByFsId(archive, String(imageArchive.fsId || ""), pullLocalRel(archive, imageArchive.remotePath));
|
||||
const ok = n8nText.ok === true && n8nImage.ok === true && textArchive.ok === true && imageArchive.ok === true && textPull.ok === true && imagePull.ok === true;
|
||||
return {
|
||||
ok,
|
||||
action: "platform-infra-wechat-archive-validate",
|
||||
config: configSummary(archive),
|
||||
fixtures: {
|
||||
text: validationSummary(textPayload, n8nText, textArchive, textPull, options.full),
|
||||
image: validationSummary(imagePayload, n8nImage, imageArchive, imagePull, options.full),
|
||||
},
|
||||
valuesPrinted: false,
|
||||
next: {
|
||||
pullText: `bun scripts/cli.ts platform-infra wechat-archive pull --remote-path ${textArchive.remotePath} --target ${archive.target.id}`,
|
||||
pullImage: `bun scripts/cli.ts platform-infra wechat-archive pull --remote-path ${imageArchive.remotePath} --target ${archive.target.id}`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function pull(options: PullOptions): Promise<Record<string, unknown>> {
|
||||
const archive = readConfig();
|
||||
assertTarget(archive, options.targetId);
|
||||
if (!options.remotePath && !options.fsId) throw new Error("pull requires --remote-path or --fs-id");
|
||||
let fsId = options.fsId;
|
||||
let remotePath = options.remotePath;
|
||||
let listed: Record<string, unknown> | null = null;
|
||||
if (!fsId) {
|
||||
remotePath = normalizeRemotePath(String(remotePath));
|
||||
listed = findBaiduFileByRemotePath(archive.baiduNetdisk.serviceId, remotePath);
|
||||
if (listed === null) throw new Error(`Baidu file not found: ${remotePath}`);
|
||||
fsId = String(listed.fsId || listed.fs_id || "");
|
||||
if (!fsId) throw new Error(`Baidu file has no fsId in listing: ${remotePath}`);
|
||||
}
|
||||
const localRel = options.localPath ?? pullLocalRel(archive, remotePath ?? `fsid-${fsId}`);
|
||||
const downloaded = await pullByFsId(archive, fsId, localRel);
|
||||
return {
|
||||
ok: downloaded.ok,
|
||||
action: "platform-infra-wechat-archive-pull",
|
||||
remotePath: remotePath ?? downloaded.remotePath,
|
||||
fsId,
|
||||
listed: redactBaiduFileEntry(listed),
|
||||
download: options.full ? downloaded : compactDownload(downloaded),
|
||||
valuesPrinted: false,
|
||||
...(options.raw ? { raw: downloaded } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function readConfig(): WechatArchiveConfig {
|
||||
const root = readYamlRecord(configFile, "platform-infra-wechat-archive");
|
||||
const metadata = recordField(root, "metadata", configLabel);
|
||||
const target = recordField(root, "target", configLabel);
|
||||
const langbot = recordField(root, "langbot", configLabel);
|
||||
const n8n = recordField(root, "n8n", configLabel);
|
||||
const n8nWorkflow = recordField(n8n, "workflow", `${configLabel}.n8n`);
|
||||
const baiduNetdisk = recordField(root, "baiduNetdisk", configLabel);
|
||||
const staging = recordField(baiduNetdisk, "staging", `${configLabel}.baiduNetdisk`);
|
||||
const archive = recordField(baiduNetdisk, "archive", `${configLabel}.baiduNetdisk`);
|
||||
const messageTypesRaw = recordField(root, "messageTypes", configLabel);
|
||||
const validateRaw = recordField(root, "validate", configLabel);
|
||||
const fixtures = recordField(validateRaw, "fixtures", `${configLabel}.validate`);
|
||||
const textFixture = recordField(fixtures, "text", `${configLabel}.validate.fixtures`);
|
||||
const imageFixture = recordField(fixtures, "image", `${configLabel}.validate.fixtures`);
|
||||
const relatedIssuesRaw = metadata.relatedIssues;
|
||||
if (!Array.isArray(relatedIssuesRaw)) throw new Error(`${configLabel}.metadata.relatedIssues must be an array`);
|
||||
const relatedIssues = relatedIssuesRaw.map((item) => Number(item));
|
||||
const messageTypes: WechatArchiveConfig["messageTypes"] = {};
|
||||
for (const [key, value] of Object.entries(messageTypesRaw)) {
|
||||
const record = asRecord(value, `${configLabel}.messageTypes.${key}`);
|
||||
messageTypes[key] = {
|
||||
enabled: booleanField(record, "enabled", `${configLabel}.messageTypes.${key}`),
|
||||
contentField: stringField(record, "contentField", `${configLabel}.messageTypes.${key}`),
|
||||
mediaMode: stringField(record, "mediaMode", `${configLabel}.messageTypes.${key}`),
|
||||
filenameField: optionalStringField(record, "filenameField", `${configLabel}.messageTypes.${key}`),
|
||||
mimeTypeField: optionalStringField(record, "mimeTypeField", `${configLabel}.messageTypes.${key}`),
|
||||
};
|
||||
}
|
||||
const config: WechatArchiveConfig = {
|
||||
version: Number(root.version),
|
||||
kind: "platform-infra-wechat-archive",
|
||||
metadata: {
|
||||
id: stringField(metadata, "id", configLabel),
|
||||
owner: stringField(metadata, "owner", configLabel),
|
||||
relatedIssues,
|
||||
},
|
||||
target: {
|
||||
id: stringField(target, "id", configLabel),
|
||||
route: stringField(target, "route", configLabel),
|
||||
namespace: stringField(target, "namespace", configLabel),
|
||||
},
|
||||
langbot: {
|
||||
configRef: stringField(langbot, "configRef", configLabel),
|
||||
publicBaseUrl: stringField(langbot, "publicBaseUrl", configLabel),
|
||||
expectedAdapter: stringField(langbot, "expectedAdapter", configLabel),
|
||||
callbackPath: stringField(langbot, "callbackPath", configLabel),
|
||||
notes: Array.isArray(langbot.notes) ? langbot.notes.map(String) : [],
|
||||
},
|
||||
n8n: {
|
||||
configRef: stringField(n8n, "configRef", configLabel),
|
||||
publicBaseUrl: stringField(n8n, "publicBaseUrl", configLabel),
|
||||
workflow: {
|
||||
name: stringField(n8nWorkflow, "name", `${configLabel}.n8n.workflow`),
|
||||
id: stringField(n8nWorkflow, "id", `${configLabel}.n8n.workflow`),
|
||||
webhookPath: stringField(n8nWorkflow, "webhookPath", `${configLabel}.n8n.workflow`),
|
||||
active: booleanField(n8nWorkflow, "active", `${configLabel}.n8n.workflow`),
|
||||
timezone: stringField(n8nWorkflow, "timezone", `${configLabel}.n8n.workflow`),
|
||||
},
|
||||
},
|
||||
baiduNetdisk: {
|
||||
serviceId: stringField(baiduNetdisk, "serviceId", `${configLabel}.baiduNetdisk`),
|
||||
proxyMode: stringField(baiduNetdisk, "proxyMode", `${configLabel}.baiduNetdisk`),
|
||||
configRef: stringField(baiduNetdisk, "configRef", `${configLabel}.baiduNetdisk`),
|
||||
staging: {
|
||||
hostRoot: stringField(staging, "hostRoot", `${configLabel}.baiduNetdisk.staging`),
|
||||
containerRoot: stringField(staging, "containerRoot", `${configLabel}.baiduNetdisk.staging`),
|
||||
outboxDir: stringField(staging, "outboxDir", `${configLabel}.baiduNetdisk.staging`),
|
||||
pullDir: stringField(staging, "pullDir", `${configLabel}.baiduNetdisk.staging`),
|
||||
},
|
||||
archive: {
|
||||
remoteRoot: normalizeRemotePath(stringField(archive, "remoteRoot", `${configLabel}.baiduNetdisk.archive`)),
|
||||
timezone: stringField(archive, "timezone", `${configLabel}.baiduNetdisk.archive`),
|
||||
textExtension: stringField(archive, "textExtension", `${configLabel}.baiduNetdisk.archive`),
|
||||
imageExtension: stringField(archive, "imageExtension", `${configLabel}.baiduNetdisk.archive`),
|
||||
pathTemplate: stringField(archive, "pathTemplate", `${configLabel}.baiduNetdisk.archive`),
|
||||
maxInlineBytes: numberField(archive, "maxInlineBytes", `${configLabel}.baiduNetdisk.archive`),
|
||||
},
|
||||
},
|
||||
messageTypes,
|
||||
validate: {
|
||||
timeoutMs: numberField(validateRaw, "timeoutMs", `${configLabel}.validate`),
|
||||
pollIntervalMs: numberField(validateRaw, "pollIntervalMs", `${configLabel}.validate`),
|
||||
fixtures: {
|
||||
text: {
|
||||
conversationId: stringField(textFixture, "conversationId", `${configLabel}.validate.fixtures.text`),
|
||||
senderId: stringField(textFixture, "senderId", `${configLabel}.validate.fixtures.text`),
|
||||
text: stringField(textFixture, "text", `${configLabel}.validate.fixtures.text`),
|
||||
},
|
||||
image: {
|
||||
conversationId: stringField(imageFixture, "conversationId", `${configLabel}.validate.fixtures.image`),
|
||||
senderId: stringField(imageFixture, "senderId", `${configLabel}.validate.fixtures.image`),
|
||||
filename: stringField(imageFixture, "filename", `${configLabel}.validate.fixtures.image`),
|
||||
mimeType: stringField(imageFixture, "mimeType", `${configLabel}.validate.fixtures.image`),
|
||||
dataBase64: stringField(imageFixture, "dataBase64", `${configLabel}.validate.fixtures.image`),
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
validateConfig(config);
|
||||
return config;
|
||||
}
|
||||
|
||||
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`);
|
||||
for (const ref of [config.langbot.configRef, config.n8n.configRef]) {
|
||||
const path = resolveRepoPath(ref);
|
||||
if (!existsSync(path)) throw new Error(`${ref} does not exist`);
|
||||
}
|
||||
if (!config.n8n.publicBaseUrl.startsWith("https://")) throw new Error(`${configLabel}.n8n.publicBaseUrl must be https`);
|
||||
if (!config.langbot.publicBaseUrl.startsWith("https://")) throw new Error(`${configLabel}.langbot.publicBaseUrl must be https`);
|
||||
if (!config.n8n.workflow.webhookPath || config.n8n.workflow.webhookPath.startsWith("/") || config.n8n.workflow.webhookPath.includes("/")) {
|
||||
throw new Error(`${configLabel}.n8n.workflow.webhookPath must be one relative path segment`);
|
||||
}
|
||||
if (!config.baiduNetdisk.archive.pathTemplate.includes("{{remoteRoot}}")) throw new Error(`${configLabel}.baiduNetdisk.archive.pathTemplate must include {{remoteRoot}}`);
|
||||
if (config.baiduNetdisk.proxyMode !== "backend-core-microservice-proxy") throw new Error(`${configLabel}.baiduNetdisk.proxyMode must be backend-core-microservice-proxy`);
|
||||
}
|
||||
|
||||
function assertTarget(config: WechatArchiveConfig, targetId: string): void {
|
||||
if (targetId !== config.target.id) throw new Error(`wechat archive target ${targetId} is not declared in ${configLabel}`);
|
||||
}
|
||||
|
||||
function configSummary(config: WechatArchiveConfig): Record<string, unknown> {
|
||||
return {
|
||||
path: configLabel,
|
||||
id: config.metadata.id,
|
||||
relatedIssues: config.metadata.relatedIssues,
|
||||
target: config.target,
|
||||
langbot: {
|
||||
configRef: config.langbot.configRef,
|
||||
publicBaseUrl: config.langbot.publicBaseUrl,
|
||||
expectedAdapter: config.langbot.expectedAdapter,
|
||||
callbackPath: config.langbot.callbackPath,
|
||||
},
|
||||
n8n: {
|
||||
configRef: config.n8n.configRef,
|
||||
publicBaseUrl: config.n8n.publicBaseUrl,
|
||||
workflow: config.n8n.workflow,
|
||||
webhookUrl: webhookUrl(config),
|
||||
},
|
||||
baiduNetdisk: {
|
||||
serviceId: config.baiduNetdisk.serviceId,
|
||||
proxyMode: config.baiduNetdisk.proxyMode,
|
||||
configRef: config.baiduNetdisk.configRef,
|
||||
remoteRoot: config.baiduNetdisk.archive.remoteRoot,
|
||||
staging: config.baiduNetdisk.staging,
|
||||
},
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function policyChecks(config: WechatArchiveConfig): Array<Record<string, unknown>> {
|
||||
return [
|
||||
{ name: "yaml-source", ok: repoRelative(configFile) === configLabel, detail: `${configLabel} is the archive workflow source of truth.` },
|
||||
{ name: "langbot-yaml-ref", ok: config.langbot.configRef === "config/platform-infra/langbot.yaml", detail: "LangBot base service remains declared in YAML." },
|
||||
{ name: "n8n-yaml-ref", ok: config.n8n.configRef === "config/platform-infra/n8n.yaml", detail: "n8n base service remains declared in YAML." },
|
||||
{ name: "baidu-private-proxy", ok: config.baiduNetdisk.proxyMode === "backend-core-microservice-proxy", detail: "Baidu Netdisk access stays behind backend-core microservice proxy." },
|
||||
{ name: "single-pk01-postgres", ok: true, detail: "LangBot and n8n use dedicated databases inside the single PK01/Pika01 external PostgreSQL instance; this workflow adds no PostgreSQL instance." },
|
||||
{ name: "text-enabled", ok: config.messageTypes.text?.enabled === true, detail: "Text messages are archive-enabled." },
|
||||
{ name: "image-enabled", ok: config.messageTypes.image?.enabled === true, detail: "Image messages are archive-enabled." },
|
||||
{ name: "no-secret-output", ok: true, detail: "Baidu, LangBot and n8n secrets are referenced by existing services and are not printed by this CLI." },
|
||||
];
|
||||
}
|
||||
|
||||
function webhookUrl(config: WechatArchiveConfig): string {
|
||||
return `${config.n8n.publicBaseUrl.replace(/\/+$/u, "")}/webhook/${registeredWebhookPath(config)}`;
|
||||
}
|
||||
|
||||
function renderN8nWorkflow(config: WechatArchiveConfig): Record<string, unknown> {
|
||||
const code = `
|
||||
const input = $input.first()?.json ?? {};
|
||||
const body = input.body && typeof input.body === 'object' ? input.body : input;
|
||||
function pick(...keys) {
|
||||
for (const key of keys) {
|
||||
const value = key.split('.').reduce((acc, part) => acc && typeof acc === 'object' ? acc[part] : undefined, body);
|
||||
if (value !== undefined && value !== null && String(value).length > 0) return value;
|
||||
}
|
||||
return '';
|
||||
}
|
||||
const messageType = String(pick('messageType', 'type', 'msgType') || 'text').toLowerCase();
|
||||
const receivedAt = String(pick('receivedAt', 'timestamp') || new Date().toISOString());
|
||||
const messageId = String(pick('messageId', 'msgId', 'id') || String(Date.now()));
|
||||
const conversationId = String(pick('conversationId', 'chatId', 'roomId', 'fromUser') || 'unknown-conversation');
|
||||
const senderId = String(pick('senderId', 'userId', 'fromUserName', 'fromUser') || 'unknown-sender');
|
||||
const normalized = {
|
||||
ok: true,
|
||||
source: 'n8n',
|
||||
workflow: '${config.n8n.workflow.id}',
|
||||
message: {
|
||||
messageId,
|
||||
conversationId,
|
||||
senderId,
|
||||
messageType,
|
||||
text: String(pick('text', 'content') || ''),
|
||||
receivedAt,
|
||||
media: body.media && typeof body.media === 'object' ? body.media : {},
|
||||
},
|
||||
archive: {
|
||||
remoteRoot: '${config.baiduNetdisk.archive.remoteRoot}',
|
||||
pathTemplate: '${config.baiduNetdisk.archive.pathTemplate}',
|
||||
},
|
||||
};
|
||||
return [{ json: normalized }];
|
||||
`.trim();
|
||||
return {
|
||||
id: config.n8n.workflow.id,
|
||||
name: config.n8n.workflow.name,
|
||||
active: config.n8n.workflow.active,
|
||||
nodes: [
|
||||
{
|
||||
parameters: {
|
||||
httpMethod: "POST",
|
||||
path: config.n8n.workflow.webhookPath,
|
||||
responseMode: "lastNode",
|
||||
options: {},
|
||||
},
|
||||
id: "webhook",
|
||||
name: "wechat-archive-webhook",
|
||||
type: "n8n-nodes-base.webhook",
|
||||
typeVersion: 2,
|
||||
position: [0, 0],
|
||||
},
|
||||
{
|
||||
parameters: { jsCode: code },
|
||||
id: "normalize",
|
||||
name: "Normalize WeChat Message",
|
||||
type: "n8n-nodes-base.code",
|
||||
typeVersion: 2,
|
||||
position: [280, 0],
|
||||
},
|
||||
],
|
||||
connections: {
|
||||
"wechat-archive-webhook": {
|
||||
main: [[{ node: "Normalize WeChat Message", type: "main", index: 0 }]],
|
||||
},
|
||||
},
|
||||
settings: {
|
||||
executionOrder: "v1",
|
||||
timezone: config.n8n.workflow.timezone,
|
||||
},
|
||||
staticData: null,
|
||||
tags: [],
|
||||
};
|
||||
}
|
||||
|
||||
function registeredWebhookPath(config: WechatArchiveConfig): string {
|
||||
return [
|
||||
config.n8n.workflow.id,
|
||||
"wechat-archive-webhook",
|
||||
config.n8n.workflow.webhookPath.replace(/^\/+|\/+$/gu, ""),
|
||||
].map((segment) => encodeURIComponent(segment)).join("/");
|
||||
}
|
||||
|
||||
function fixturePayload(config: WechatArchiveConfig, type: "text" | "image", observedAt: Date): Record<string, unknown> {
|
||||
const suffix = randomUUID().slice(0, 8);
|
||||
if (type === "text") {
|
||||
const fixture = config.validate.fixtures.text;
|
||||
return {
|
||||
source: "unidesk-cli-validate",
|
||||
messageType: "text",
|
||||
messageId: `validate-text-${suffix}`,
|
||||
conversationId: fixture.conversationId,
|
||||
senderId: fixture.senderId,
|
||||
text: `${fixture.text} ${observedAt.toISOString()}`,
|
||||
receivedAt: observedAt.toISOString(),
|
||||
};
|
||||
}
|
||||
const fixture = config.validate.fixtures.image;
|
||||
return {
|
||||
source: "unidesk-cli-validate",
|
||||
messageType: "image",
|
||||
messageId: `validate-image-${suffix}`,
|
||||
conversationId: fixture.conversationId,
|
||||
senderId: fixture.senderId,
|
||||
receivedAt: observedAt.toISOString(),
|
||||
media: {
|
||||
filename: fixture.filename,
|
||||
mimeType: fixture.mimeType,
|
||||
dataBase64: fixture.dataBase64,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function callWorkflow(config: WechatArchiveConfig, payload: Record<string, unknown>): Promise<Record<string, unknown>> {
|
||||
const response = await fetchJsonWithTimeout(webhookUrl(config), payload, 45_000);
|
||||
const body = typeof response.body === "object" && response.body !== null && !Array.isArray(response.body)
|
||||
? response.body as Record<string, unknown>
|
||||
: {};
|
||||
return {
|
||||
ok: response.ok && (body.ok === true || response.status >= 200 && response.status < 300),
|
||||
status: response.status,
|
||||
webhookUrl: webhookUrl(config),
|
||||
body,
|
||||
};
|
||||
}
|
||||
|
||||
async function archiveMessage(config: WechatArchiveConfig, originalPayload: Record<string, unknown>, n8nResponse: Record<string, unknown>): Promise<Record<string, unknown>> {
|
||||
if (n8nResponse.ok !== true) {
|
||||
return {
|
||||
ok: false,
|
||||
error: "n8n-webhook-failed",
|
||||
n8n: {
|
||||
status: n8nResponse.status,
|
||||
body: compactUnknown(n8nResponse.body),
|
||||
},
|
||||
};
|
||||
}
|
||||
const body = asRecord(n8nResponse.body ?? {}, "n8n response body");
|
||||
const normalized = asRecord(body.message ?? originalPayload, "normalized message");
|
||||
const type = String(normalized.messageType ?? originalPayload.messageType ?? "text").toLowerCase();
|
||||
const messageId = sanitizePathSegment(String(normalized.messageId ?? originalPayload.messageId ?? randomUUID()));
|
||||
const conversationId = sanitizePathSegment(String(normalized.conversationId ?? originalPayload.conversationId ?? "unknown-conversation"));
|
||||
const senderId = sanitizePathSegment(String(normalized.senderId ?? originalPayload.senderId ?? "unknown-sender"));
|
||||
const receivedAt = new Date(String(normalized.receivedAt ?? originalPayload.receivedAt ?? new Date().toISOString()));
|
||||
const extension = type === "image" ? config.baiduNetdisk.archive.imageExtension : config.baiduNetdisk.archive.textExtension;
|
||||
const remotePath = normalizeRemotePath(renderTemplate(config.baiduNetdisk.archive.pathTemplate, {
|
||||
remoteRoot: config.baiduNetdisk.archive.remoteRoot,
|
||||
date: dateInTimeZone(receivedAt, config.baiduNetdisk.archive.timezone),
|
||||
timestamp: receivedAt.toISOString().replace(/[-:.TZ]/gu, ""),
|
||||
conversationId,
|
||||
senderId,
|
||||
messageId,
|
||||
type,
|
||||
extension,
|
||||
filename: `${messageId}.${extension}`,
|
||||
}));
|
||||
const localRel = relativeStagingPath(config.baiduNetdisk.staging.outboxDir, `${messageId}.${extension}`);
|
||||
const written = type === "image"
|
||||
? writeStagingBase64({ hostRoot: config.baiduNetdisk.staging.hostRoot, relativePath: localRel, dataBase64: String(asRecord(normalized.media ?? originalPayload.media ?? {}, "image media").dataBase64 ?? "") })
|
||||
: writeStagingText({ hostRoot: config.baiduNetdisk.staging.hostRoot, relativePath: localRel, content: messageText(normalized, originalPayload) });
|
||||
const created = assertProxyOk(microserviceProxy(config.baiduNetdisk.serviceId, "/api/transfers/upload-from-path", {
|
||||
method: "POST",
|
||||
body: { localPath: localRel, remotePath },
|
||||
timeoutMs: 60_000,
|
||||
}), "baidu upload create");
|
||||
const job = asRecord(created.job, "baidu upload job");
|
||||
const jobId = String(job.id || "");
|
||||
const waited = await waitForBaiduTransfer(config.baiduNetdisk.serviceId, jobId, {
|
||||
timeoutMs: config.validate.timeoutMs,
|
||||
pollIntervalMs: config.validate.pollIntervalMs,
|
||||
});
|
||||
const waitedJob = asRecord(waited.job ?? {}, "baidu waited upload job");
|
||||
const result = asRecord(waitedJob.result ?? {}, "baidu upload result");
|
||||
const baiduResult = asRecord(result.baidu ?? {}, "baidu upload result.baidu");
|
||||
let fsId = String(baiduResult.fs_id ?? baiduResult.fsId ?? waitedJob.fsId ?? "");
|
||||
let listed: Record<string, unknown> | null = null;
|
||||
if (!fsId) {
|
||||
listed = findBaiduFileByRemotePath(config.baiduNetdisk.serviceId, remotePath);
|
||||
fsId = String(listed?.fsId ?? listed?.fs_id ?? "");
|
||||
}
|
||||
return {
|
||||
ok: waited.ok === true && Boolean(fsId),
|
||||
type,
|
||||
remotePath,
|
||||
fsId,
|
||||
messageId,
|
||||
conversationId,
|
||||
senderId,
|
||||
local: { relativePath: localRel, hostPath: written.hostPath, bytes: written.bytes, sha256: written.sha256 },
|
||||
uploadJob: waitedJob,
|
||||
listed: redactBaiduFileEntry(listed),
|
||||
};
|
||||
}
|
||||
|
||||
async function pullByFsId(config: WechatArchiveConfig, fsId: string, localRel: string): Promise<Record<string, unknown>> {
|
||||
if (!fsId) return { ok: false, error: "fsId is required" };
|
||||
const created = assertProxyOk(microserviceProxy(config.baiduNetdisk.serviceId, "/api/transfers/download-to-path", {
|
||||
method: "POST",
|
||||
body: { fsId, localPath: localRel },
|
||||
timeoutMs: 60_000,
|
||||
}), "baidu download create");
|
||||
const job = asRecord(created.job, "baidu download job");
|
||||
const jobId = String(job.id || "");
|
||||
const waited = await waitForBaiduTransfer(config.baiduNetdisk.serviceId, jobId, {
|
||||
timeoutMs: config.validate.timeoutMs,
|
||||
pollIntervalMs: config.validate.pollIntervalMs,
|
||||
});
|
||||
const waitedJob = asRecord(waited.job ?? {}, "baidu waited download job");
|
||||
const containerPath = String(waitedJob.localPath || "");
|
||||
const hostPath = containerPathToHostPath(config.baiduNetdisk.staging.containerRoot, config.baiduNetdisk.staging.hostRoot, containerPath)
|
||||
?? hostRootPath(pathPosix.join(config.baiduNetdisk.staging.hostRoot, localRel));
|
||||
return {
|
||||
ok: waited.ok === true && existsSync(hostPath),
|
||||
fsId,
|
||||
remotePath: waitedJob.remotePath,
|
||||
localPath: containerPath,
|
||||
hostPath,
|
||||
sha256: existsSync(hostPath) ? sha256File(hostPath) : null,
|
||||
downloadJob: waitedJob,
|
||||
};
|
||||
}
|
||||
|
||||
function messageText(normalized: Record<string, unknown>, originalPayload: Record<string, unknown>): string {
|
||||
const text = String(normalized.text ?? originalPayload.text ?? originalPayload.content ?? "");
|
||||
const payloadHash = createHash("sha256").update(JSON.stringify(originalPayload)).digest("hex");
|
||||
return `${text}\n\n---\nsource=unidesk-wechat-archive\npayloadSha256=${payloadHash}\n`;
|
||||
}
|
||||
|
||||
function pullLocalRel(config: WechatArchiveConfig, remotePath: unknown): string {
|
||||
const name = sanitizePathSegment(pathPosix.basename(String(remotePath || `pull-${randomUUID()}`)), `pull-${randomUUID()}`);
|
||||
return pathPosix.join(config.baiduNetdisk.staging.pullDir, name);
|
||||
}
|
||||
|
||||
function validationSummary(
|
||||
payload: Record<string, unknown>,
|
||||
n8n: Record<string, unknown>,
|
||||
archive: Record<string, unknown>,
|
||||
downloaded: Record<string, unknown>,
|
||||
full: boolean,
|
||||
): Record<string, unknown> {
|
||||
return {
|
||||
ok: n8n.ok === true && archive.ok === true && downloaded.ok === true,
|
||||
messageId: payload.messageId,
|
||||
n8n: full ? n8n : { ok: n8n.ok, status: n8n.status, webhookUrl: n8n.webhookUrl },
|
||||
archive: full ? archive : {
|
||||
ok: archive.ok,
|
||||
remotePath: archive.remotePath,
|
||||
fsId: archive.fsId,
|
||||
local: archive.local,
|
||||
},
|
||||
pull: full ? downloaded : compactDownload(downloaded),
|
||||
};
|
||||
}
|
||||
|
||||
function compactDownload(downloaded: Record<string, unknown>): Record<string, unknown> {
|
||||
return {
|
||||
ok: downloaded.ok,
|
||||
fsId: downloaded.fsId,
|
||||
remotePath: downloaded.remotePath,
|
||||
hostPath: downloaded.hostPath,
|
||||
sha256: downloaded.sha256,
|
||||
};
|
||||
}
|
||||
|
||||
function redactBaiduFileEntry(value: Record<string, unknown> | null): Record<string, unknown> | null {
|
||||
if (value === null) return null;
|
||||
const copy: Record<string, unknown> = { ...value };
|
||||
if (copy.thumbs !== undefined) copy.thumbs = "<redacted-signed-thumbnail-urls>";
|
||||
if (copy.dlink !== undefined) copy.dlink = "<redacted-download-link>";
|
||||
return copy;
|
||||
}
|
||||
@@ -191,7 +191,7 @@ interface EgressProxySecretMaterial {
|
||||
|
||||
export function platformInfraHelp(): unknown {
|
||||
return {
|
||||
command: "platform-infra sub2api|langbot|n8n ...",
|
||||
command: "platform-infra sub2api|langbot|n8n|wechat-archive ...",
|
||||
output: "json",
|
||||
usage: [
|
||||
"bun scripts/cli.ts platform-infra sub2api plan [--target G14|D601]",
|
||||
@@ -216,8 +216,13 @@ export function platformInfraHelp(): unknown {
|
||||
"bun scripts/cli.ts platform-infra n8n status [--target G14] [--full|--raw]",
|
||||
"bun scripts/cli.ts platform-infra n8n logs [--target G14] [--component app|frpc|all]",
|
||||
"bun scripts/cli.ts platform-infra n8n validate [--target G14] [--full|--raw]",
|
||||
"bun scripts/cli.ts platform-infra wechat-archive plan [--target G14]",
|
||||
"bun scripts/cli.ts platform-infra wechat-archive apply [--target G14] --confirm",
|
||||
"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/...",
|
||||
],
|
||||
description: "Operate YAML-controlled platform-infra services such as Sub2API, LangBot and n8n. Public services use PK01 Caddy+FRP rather than Kubernetes Ingress, NodePort, or LoadBalancer.",
|
||||
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: {
|
||||
default: defaultTargetId,
|
||||
namespace,
|
||||
@@ -250,6 +255,10 @@ export async function runPlatformInfraCommand(config: UniDeskConfig, args: strin
|
||||
const { runN8nCommand } = await import("./platform-infra-n8n");
|
||||
return await runN8nCommand(config, args.slice(1));
|
||||
}
|
||||
if (target === "wechat-archive") {
|
||||
const { runWechatArchiveCommand } = await import("./platform-infra-wechat-archive");
|
||||
return await runWechatArchiveCommand(config, args.slice(1));
|
||||
}
|
||||
if (target !== "sub2api") return unsupported(args);
|
||||
if (action === "plan" || action === undefined) return plan(parseTargetOptions(args.slice(2)));
|
||||
if (action === "apply") return await apply(config, parseApplyOptions(args.slice(2)));
|
||||
|
||||
Reference in New Issue
Block a user