#!/usr/bin/env python3 """Upload Single Window declaration XML/ZIP files without changing the originals.""" import argparse import hashlib import json import logging import mimetypes import os import sqlite3 import sys import time import urllib.error import urllib.parse import urllib.request import uuid from pathlib import Path AGENT_VERSION = "2026.11.09.3" def load_config(path): with path.open("r", encoding="utf-8-sig") as handle: config = json.load(handle) required = ["server_url", "source_id", "token", "watch_dirs"] missing = [key for key in required if not config.get(key)] if missing: raise ValueError("配置缺少：" + "、".join(missing)) if not isinstance(config["watch_dirs"], list): raise ValueError("watch_dirs 必须是目录数组") server_url = str(config["server_url"]).strip().rstrip("/") if not server_url.startswith("https://") and not config.get("allow_http_for_local_test"): raise ValueError("server_url 必须使用 https://") config["server_url"] = server_url return config def open_state(path): path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(str(path)) conn.execute( """CREATE TABLE IF NOT EXISTS uploaded_files ( content_hash TEXT PRIMARY KEY, file_path TEXT, file_size INTEGER, modified_at REAL, status TEXT, attempted_at REAL, response_text TEXT )""" ) conn.commit() return conn def request_json(url, source_id, token, method="GET", body=None, content_type="application/json"): request = urllib.request.Request(url, data=body, method=method) request.add_header("Authorization", "Bearer " + token) request.add_header("X-Customs-Sync-Source", source_id) if body is not None: request.add_header("Content-Type", content_type) try: with urllib.request.urlopen(request, timeout=60) as response: raw = response.read().decode("utf-8", errors="replace") return json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: detail = json.loads(raw).get("detail") or raw except Exception: detail = raw raise RuntimeError("HTTP {}：{}".format(exc.code, detail)) from exc def multipart_file(path): boundary = "----CustomsSync" + uuid.uuid4().hex content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream" body = bytearray() body.extend(("--{}\r\n".format(boundary)).encode("ascii")) safe_name = path.name.replace('"', "_") body.extend(("Content-Disposition: form-data; name=\"file\"; filename=\"{}\"\r\n".format(safe_name)).encode("utf-8")) body.extend(("Content-Type: {}\r\n\r\n".format(content_type)).encode("ascii")) body.extend(path.read_bytes()) body.extend(("\r\n--{}--\r\n".format(boundary)).encode("ascii")) return bytes(body), "multipart/form-data; boundary=" + boundary def sha256_file(path): digest = hashlib.sha256() with path.open("rb") as handle: while True: chunk = handle.read(1024 * 1024) if not chunk: break digest.update(chunk) return digest.hexdigest() def candidate_files(config): max_bytes = int(config.get("max_file_mb", 50)) * 1024 * 1024 for raw_dir in config["watch_dirs"]: folder = Path(os.path.expandvars(os.path.expanduser(str(raw_dir)))) if not folder.is_dir(): logging.warning("监控目录不存在：%s", folder) continue for path in folder.rglob("*"): try: if not path.is_file() or path.suffix.lower() not in {".xml", ".zip"}: continue stat = path.stat() if stat.st_size <= 0 or stat.st_size > max_bytes: continue yield path, stat except OSError as exc: logging.warning("读取文件信息失败：%s：%s", path, exc) def upload_ready_files(config, conn): stable_seconds = int(config.get("stable_seconds", 10)) retry_seconds = int(config.get("retry_seconds", 300)) now = time.time() uploaded = 0 for path, stat in candidate_files(config): if now - stat.st_mtime < stable_seconds: continue try: content_hash = sha256_file(path) except OSError as exc: logging.warning("读取文件失败：%s：%s", path, exc) continue previous = conn.execute( "SELECT status,attempted_at FROM uploaded_files WHERE content_hash=?", (content_hash,) ).fetchone() if previous and previous[0] == "success": continue if previous and previous[0] == "error" and now - float(previous[1] or 0) < retry_seconds: continue logging.info("上传：%s", path) try: body, content_type = multipart_file(path) result = request_json( config["server_url"] + "/api/customs-declaration-sync/upload", config["source_id"], config["token"], "POST", body, content_type, ) conn.execute( "INSERT OR REPLACE INTO uploaded_files VALUES (?,?,?,?,?,?,?)", (content_hash, str(path), stat.st_size, stat.st_mtime, "success", time.time(), json.dumps(result, ensure_ascii=False)[:10000]), ) conn.commit() uploaded += 1 logging.info("完成：%s（%s）", path.name, result.get("status") or "duplicate") except Exception as exc: conn.execute( "INSERT OR REPLACE INTO uploaded_files VALUES (?,?,?,?,?,?,?)", (content_hash, str(path), stat.st_size, stat.st_mtime, "error", time.time(), str(exc)[:10000]), ) conn.commit() logging.error("上传失败：%s：%s", path, exc) return uploaded DOC_NAMES = { "customs_agreement": "委托报关协议", "paperless_release_notice": "无纸化放行通知书", } def document_api_request(config, path, method="GET", body=None, content_type="application/json"): return request_json( config["server_url"] + path, config["source_id"], config["token"], method, body, content_type, ) def multipart_document(declaration_id, doc_type, path): boundary = "----SingleWindowDocument" + uuid.uuid4().hex body = bytearray() def field(name, value): body.extend(("--{}\r\n".format(boundary)).encode("ascii")) body.extend(("Content-Disposition: form-data; name=\"{}\"\r\n\r\n{}\r\n".format(name, value)).encode("utf-8")) field("declarationId", str(declaration_id)) field("docType", doc_type) safe_name = path.name.replace('"', "_") mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream" body.extend(("--{}\r\n".format(boundary)).encode("ascii")) body.extend(("Content-Disposition: form-data; name=\"file\"; filename=\"{}\"\r\n".format(safe_name)).encode("utf-8")) body.extend(("Content-Type: {}\r\n\r\n".format(mime)).encode("ascii")) body.extend(path.read_bytes()) body.extend(("\r\n--{}--\r\n".format(boundary)).encode("ascii")) return bytes(body), "multipart/form-data; boundary=" + boundary def upload_document(config, declaration_id, doc_type, path): body, content_type = multipart_document(declaration_id, doc_type, path) return document_api_request( config, "/api/customs-declaration-sync/supporting-document-upload", "POST", body, content_type, ) def frame_locators(page): return list(page.frames) def visible(locator): try: return locator.count() > 0 and locator.first.is_visible() except Exception: return False def first_actionable(locator): try: for index in range(locator.count()): candidate = locator.nth(index) if candidate.is_visible() and candidate.is_enabled(): return candidate except Exception: pass return None def find_text(page, text, exact=True): for frame in frame_locators(page): for locator in ( frame.get_by_role("button", name=text, exact=exact), frame.get_by_text(text, exact=exact), ): candidate = first_actionable(locator) if candidate is not None: return candidate return None def click_text(page, text, exact=True): locator = find_text(page, text, exact) if locator is None: raise RuntimeError("未找到页面控件：" + text) locator.click() def find_option_in_table_row(page, row_text, option_text): for frame in frame_locators(page): rows = frame.get_by_text(row_text, exact=True) try: for index in range(rows.count()): row_label = rows.nth(index) if not row_label.is_visible(): continue table_row = row_label.locator("xpath=ancestor::tr[1]") if not table_row.count(): continue option = first_actionable(table_row.get_by_text(option_text, exact=True)) if option is not None: return option except Exception: continue return None def option_is_checked(option): try: return bool(option.evaluate( """element => { const control = element.matches('input[type="checkbox"], input[type="radio"]') ? element : (element.control || (element.htmlFor ? document.getElementById(element.htmlFor) : null)); return Boolean(control && control.checked); }""" )) except Exception: return False def find_customs_input(page, configured_selector=""): for frame in frame_locators(page): if configured_selector: locator = frame.locator(configured_selector) if visible(locator): return locator.first for label in ("海关编号", "18位报关单号", "报关单号"): locator = frame.get_by_label(label, exact=False) if visible(locator): return locator.first label_node = frame.get_by_text(label, exact=True) if visible(label_node): nearby = label_node.first.locator("xpath=following::input[1]") if visible(nearby): return nearby.first for selector in ( "input[placeholder*='海关编号']", "input[placeholder*='报关单号']", "input[name*='entryId' i]", "input[id*='entryId' i]", ): locator = frame.locator(selector) if visible(locator): return locator.first return None def find_declaration_query_page(context, configured_selector=""): """Return the newest open tab containing the declaration-number input.""" for candidate in reversed(context.pages): try: if candidate.is_closed(): continue if find_customs_input(candidate, configured_selector) is not None: return candidate except Exception: continue return None def select_result_row(page, customs_no): for frame in frame_locators(page): cell = frame.get_by_text(customs_no, exact=False) if not visible(cell): continue row = cell.first.locator("xpath=ancestor::tr[1]") if row.count(): checkbox = row.locator("input[type=checkbox]") if checkbox.count(): if not checkbox.first.is_checked(): checkbox.first.check(force=True) return row.click() return raise RuntimeError("查询结果中未找到报关单：" + customs_no) def directory_state(folder): state = {} if not folder.is_dir(): return state for path in folder.iterdir(): try: if path.is_file(): state[path.resolve()] = (path.stat().st_size, path.stat().st_mtime_ns) except OSError: pass return state def wait_new_document(folder, before, timeout): deadline = time.time() + timeout stable = {} while time.time() < deadline: path = find_stable_new_document(folder, before, stable) if path is not None: return path time.sleep(1) raise RuntimeError("等待单一窗口生成PDF/OFD超时") def find_stable_new_document(folder, before, stable): current = directory_state(folder) candidates = [ path for path, signature in current.items() if path.suffix.lower() in {".pdf", ".ofd"} and before.get(path) != signature ] for path in sorted(candidates, key=lambda item: current[item][1], reverse=True): signature = current[path] if stable.get(path) == signature and signature[0] > 0: return path stable[path] = signature return None def save_failure_artifacts(page, folder, customs_no, doc_type): folder.mkdir(parents=True, exist_ok=True) prefix = "{}-{}-{}".format(time.strftime("%Y%m%d-%H%M%S"), customs_no, doc_type) try: page.screenshot(path=str(folder / (prefix + ".png")), full_page=True) except Exception: pass def query_declaration(page, customs_no, selectors): input_node = find_customs_input(page, str(selectors.get("customs_no_input") or "")) if input_node is None: raise RuntimeError("未找到海关编号输入框") input_node.fill(customs_no) click_text(page, str(selectors.get("query_button_text") or "查询")) page.wait_for_timeout(int(selectors.get("query_wait_ms") or 2000)) select_result_row(page, customs_no) def trigger_document(page, doc_type, selectors): if doc_type == "customs_agreement": click_text(page, str(selectors.get("agreement_button_text") or "打印电子委托协议"), exact=False) logging.info("已打开委托协议打印设置") return click_text(page, str(selectors.get("notice_button_text") or "打印通知书"), exact=False) logging.info("已打开回执通知书打印设置") page.wait_for_timeout(500) release_text = str(selectors.get("release_option_text") or "通关无纸化放行通知书") release_section = str(selectors.get("release_section_text") or "完整/全项申报数据打印") release_option = find_option_in_table_row(page, release_section, release_text) if release_option is None: raise RuntimeError( "未找到{}区域内可用的{}，可能尚未生成放行回执".format(release_section, release_text) ) if option_is_checked(release_option): logging.info("%s区域内的%s已默认选中", release_section, release_text) else: release_option.click() if not option_is_checked(release_option): raise RuntimeError("无法选中{}区域内的{}".format(release_section, release_text)) logging.info("已选择%s区域内的%s", release_section, release_text) def close_print_dialog(page): try: cancel_button = find_text(page, "取消", exact=True) if cancel_button is not None: cancel_button.click(timeout=3000) logging.info("已关闭未完成的打印弹窗") except Exception: pass def safe_document_filename(value): return "".join(char if char.isalnum() or char in "-_." else "_" for char in str(value)) def find_new_page(context, existing_pages): for candidate in reversed(context.pages): if candidate not in existing_pages and not candidate.is_closed(): return candidate return None def pdf_url_from_page(candidate): try: page_url = str(candidate.url or "") decoded_url = urllib.parse.unquote(page_url) if ".pdf" in decoded_url.lower(): return page_url title = str(candidate.title() or "") if ".pdf" in title.lower() and page_url.startswith(("http://", "https://")): return page_url resource_urls = candidate.evaluate( """() => { const values = [location.href]; for (const node of document.querySelectorAll( 'embed[type="application/pdf"], embed[src], iframe[src], object[data]' )) { values.push(node.src || node.data || ''); } for (const entry of performance.getEntriesByType('resource')) { values.push(entry.name || ''); } return values.filter(Boolean).reverse(); }""" ) for value in resource_urls or []: resource_url = str(value or "") if ".pdf" in urllib.parse.unquote(resource_url).lower(): return resource_url except Exception: pass return "" def is_pdf_page(candidate): return bool(pdf_url_from_page(candidate)) def page_debug_summary(candidate): try: return "{} | {}".format(str(candidate.title() or "")[:80], str(candidate.url or "")[:180]) except Exception: return "标签页状态读取失败" def close_stale_pdf_pages(context, protected_pages=()): protected = set(protected_pages) closed = 0 for candidate in list(context.pages): try: if candidate in protected or candidate.is_closed() or not is_pdf_page(candidate): continue candidate.close() closed += 1 except Exception: pass if closed: logging.info("已清理%s个上一轮遗留的PDF标签页", closed) def find_preview_button(context, preferred_page, text): candidates = [preferred_page] + [item for item in reversed(context.pages) if item is not preferred_page] for candidate in candidates: try: if candidate.is_closed() or is_pdf_page(candidate): continue locator = find_text(candidate, text, exact=False) if locator is not None: return candidate, locator except Exception: continue return None, None def save_pdf_payload(download_dir, customs_no, doc_type, payload): if not payload.startswith(b"%PDF-"): raise RuntimeError("打印预览返回的内容不是PDF") filename = "{}_{}.pdf".format(customs_no, safe_document_filename(DOC_NAMES[doc_type])) path = (download_dir / filename).resolve() path.write_bytes(payload) return path def fetch_pdf_url(context, url, download_dir, customs_no, doc_type, timeout): response = context.request.get(url, timeout=timeout * 1000) if not response.ok: raise RuntimeError("下载预览PDF失败：HTTP {}".format(response.status)) return save_pdf_payload(download_dir, customs_no, doc_type, response.body()) def wait_for_preview_result( context, files_before, captured_responses, captured_downloads, download_dir, customs_no, doc_type, timeout, ): deadline = time.time() + timeout stable_files = {} next_wait_log = time.time() + 10 last_pdf_error = None while time.time() < deadline: for response in list(captured_responses): try: payload = response.body() if payload.startswith(b"%PDF-"): logging.info("已从打印预览网络响应获取PDF") return save_pdf_payload(download_dir, customs_no, doc_type, payload), None except Exception: pass for download in list(captured_downloads): try: downloaded_path = download.path() if downloaded_path: payload = Path(downloaded_path).read_bytes() if payload.startswith(b"%PDF-"): logging.info("已从浏览器下载事件获取PDF") return save_pdf_payload(download_dir, customs_no, doc_type, payload), None except Exception: pass for candidate in reversed(context.pages): try: if candidate.is_closed(): continue pdf_url = pdf_url_from_page(candidate) if pdf_url: path = fetch_pdf_url( context, pdf_url, download_dir, customs_no, doc_type, timeout ) logging.info("已识别打印预览PDF标签页") return path, candidate except Exception as exc: last_pdf_error = exc continue downloaded = find_stable_new_document(download_dir, files_before, stable_files) if downloaded is not None: payload = downloaded.read_bytes() if payload.startswith(b"%PDF-"): logging.info("已从浏览器下载目录获取PDF") return downloaded, None if time.time() >= next_wait_log: summaries = [ page_debug_summary(candidate) for candidate in context.pages if not candidate.is_closed() ] logging.info("仍在等待PDF，当前标签页：%s", "；".join(summaries) or "无") next_wait_log = time.time() + 15 time.sleep(0.25) raise RuntimeError( "点击打印预览后未捕获到PDF（当前标签页{}个、PDF响应{}个、下载事件{}个{}）".format( len(context.pages), len(captured_responses), len(captured_downloads), "、最后读取错误：{}".format(last_pdf_error) if last_pdf_error else "", ) ) def fetch_preview_document(context, page, doc_type, selectors, download_dir, customs_no, timeout): close_stale_pdf_pages(context, (page,)) captured_responses = [] captured_downloads = [] download_pages = [] def capture_pdf_response(response): try: content_type = str(response.headers.get("content-type") or "").lower() disposition = str(response.headers.get("content-disposition") or "").lower() response_url = urllib.parse.unquote(str(response.url or "")).lower() is_print_download = any(marker in response_url for marker in ("print", "preview", "download")) if ( ".pdf" in response_url or ".pdf" in disposition or "application/pdf" in content_type or ("application/octet-stream" in content_type and is_print_download) ): captured_responses.append(response) except Exception: pass def capture_download(download): captured_downloads.append(download) def attach_download_listener(target_page): try: target_page.on("download", capture_download) download_pages.append(target_page) except Exception: pass context.on("response", capture_pdf_response) context.on("page", attach_download_listener) for open_page in context.pages: attach_download_listener(open_page) existing_pages = set(context.pages) preview_page = None settings_page = page try: trigger_document(page, doc_type, selectors) direct_deadline = time.time() + 3 while time.time() < direct_deadline: candidate = find_new_page(context, existing_pages) if candidate is not None: if is_pdf_page(candidate): preview_page = candidate else: settings_page = candidate break time.sleep(0.25) if preview_page is not None: return fetch_pdf_url( context, preview_page.url, download_dir, customs_no, doc_type, timeout ) preview_text = str(selectors.get("preview_button_text") or "打印预览") deadline = time.time() + min(timeout, 30) preview_button_page = None preview_button = None while time.time() < deadline: preview_button_page, preview_button = find_preview_button(context, settings_page, preview_text) if preview_button is not None: break time.sleep(0.5) if preview_button is None: raise RuntimeError("未找到页面控件：" + preview_text) captured_responses.clear() captured_downloads.clear() close_stale_pdf_pages(context, (page, preview_button_page)) files_before_preview = directory_state(download_dir) logging.info("已找到可见的打印预览按钮，准备点击") preview_button.click() logging.info("已点击打印预览，等待PDF标签页、网络响应或下载文件") path, preview_page = wait_for_preview_result( context, files_before_preview, captured_responses, captured_downloads, download_dir, customs_no, doc_type, timeout, ) return path finally: try: context.remove_listener("response", capture_pdf_response) context.remove_listener("page", attach_download_listener) for download_page in download_pages: try: download_page.remove_listener("download", capture_download) except Exception: pass except Exception: pass try: if preview_page is not None and not preview_page.is_closed(): preview_page.close() if settings_page is not page and settings_page is not preview_page and not settings_page.is_closed(): settings_page.close() page.bring_to_front() except Exception: pass def collect_supporting_documents(config, limit=0): try: from playwright.sync_api import sync_playwright except ImportError as exc: raise RuntimeError("请先执行：pip install playwright") from exc task_result = document_api_request(config, "/api/customs-declaration-sync/supporting-document-tasks") tasks = list(task_result.get("data") or []) if limit > 0: tasks = tasks[:limit] if not tasks: logging.info("当前主体没有缺失的委托协议或放行通知书") return {"tasks": 0, "uploaded": 0, "failed": 0} download_dir = Path(os.path.expandvars(os.path.expanduser(str(config.get("download_dir") or "~/Downloads")))).resolve() download_dir.mkdir(parents=True, exist_ok=True) artifacts_dir = Path(os.path.expandvars(os.path.expanduser(str(config.get("failure_artifacts_dir") or "./single-window-failures")))).resolve() selectors = config.get("selectors") if isinstance(config.get("selectors"), dict) else {} timeout = int(config.get("document_timeout_seconds") or 90) uploaded = 0 failed = 0 with sync_playwright() as playwright: profile_dir = Path(os.path.expandvars(os.path.expanduser(str(config.get("browser_profile_dir") or "./single-window-browser-profile")))).resolve() context = playwright.chromium.launch_persistent_context( str(profile_dir), channel=str(config.get("browser_channel") or "msedge"), headless=False, accept_downloads=True, downloads_path=str(download_dir), args=["--start-maximized"], viewport=None, ) page = context.pages[0] if context.pages else context.new_page() page.goto(str(config["single_window_url"]), wait_until="domcontentloaded", timeout=120000) logging.info("请在打开的官方页面完成操作员卡登录；程序会等待报关单查询页") login_deadline = time.time() + int(config.get("login_timeout_seconds") or 600) next_wait_log = time.time() query_selector = str(selectors.get("customs_no_input") or "") while True: matched_page = find_declaration_query_page(context, query_selector) if matched_page is not None: page = matched_page try: page.bring_to_front() except Exception: pass logging.info("已识别报关数据查询页，开始处理缺失单证") break if time.time() >= login_deadline: context.close() raise RuntimeError("等待登录或报关单查询页超时") if time.time() >= next_wait_log: logging.info("仍在等待报关数据查询页，当前检测到%s个浏览器标签", len(context.pages)) next_wait_log = time.time() + 15 time.sleep(1) for task in tasks: customs_no = str(task.get("customsNo") or "") declaration_selected = False for doc_type in task.get("missingTypes") or []: if doc_type not in DOC_NAMES: continue completed = False last_error = None for attempt in range(1, 3): try: if attempt == 1: logging.info("采集：%s %s", customs_no, DOC_NAMES[doc_type]) else: logging.info("自动重试：%s %s", customs_no, DOC_NAMES[doc_type]) page.wait_for_timeout(1000) if not declaration_selected: query_declaration(page, customs_no, selectors) declaration_selected = True elif attempt == 2: logging.info("保留当前报关单选择，直接重试打印") path = fetch_preview_document( context, page, doc_type, selectors, download_dir, customs_no, timeout ) result = upload_document(config, int(task["declarationId"]), doc_type, path) logging.info( "完成：%s %s：%s", customs_no, DOC_NAMES[doc_type], result.get("msg") or "已上传", ) uploaded += 0 if result.get("skipped") else 1 completed = True break except Exception as exc: last_error = exc close_print_dialog(page) if attempt == 1: logging.warning( "首次采集未完成，将自动重试：%s %s：%s", customs_no, DOC_NAMES[doc_type], exc, ) if not completed: failed += 1 logging.error("失败：%s %s：%s", customs_no, DOC_NAMES[doc_type], last_error) save_failure_artifacts(page, artifacts_dir, customs_no, doc_type) close_print_dialog(page) if not config.get("continue_on_error", True): context.close() raise last_error context.close() return {"tasks": len(tasks), "uploaded": uploaded, "failed": failed} def main(): parser = argparse.ArgumentParser(description="单一窗口报关单同步采集程序") parser.add_argument("--config", default="customs-declaration-sync-agent.json") parser.add_argument("--once", action="store_true", help="扫描一次后退出") parser.add_argument("--test", action="store_true", help="只测试服务器连接") parser.add_argument("--collect-documents", action="store_true", help="同时补齐缺失的委托协议和放行通知书") parser.add_argument("--documents-only", action="store_true", help="只执行缺失单证补采后退出") parser.add_argument("--limit", type=int, default=0, help="单证补采最多处理多少张报关单，0为全部") args = parser.parse_args() config_path = Path(args.config).resolve() config = load_config(config_path) log_path = config_path.parent / str(config.get("log_file", "customs-declaration-sync-agent.log")) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[logging.FileHandler(str(log_path), encoding="utf-8"), logging.StreamHandler(sys.stdout)], ) logging.info("采集程序版本：%s", AGENT_VERSION) result = request_json( config["server_url"] + "/api/customs-declaration-sync/ping", config["source_id"], config["token"], ) logging.info("连接成功：%s", result.get("msg") or "同步通道可用") if args.test: return 0 if args.collect_documents or args.documents_only: if not config.get("single_window_url"): raise ValueError("单证采集需在同一份配置中填写 single_window_url") document_result = collect_supporting_documents(config, args.limit) logging.info( "单证采集结束：任务%s，上传%s，失败%s", document_result["tasks"], document_result["uploaded"], document_result["failed"], ) if args.documents_only: return 1 if document_result["failed"] else 0 state_path = config_path.parent / str(config.get("state_file", "customs-declaration-sync-agent.db")) conn = open_state(state_path) interval = max(10, int(config.get("poll_seconds", 30))) try: while True: upload_ready_files(config, conn) if args.once: break time.sleep(interval) except KeyboardInterrupt: logging.info("采集程序已停止") finally: conn.close() return 0 if __name__ == "__main__": raise SystemExit(main())