#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 1. 将二级分类图标模拟真实后台上传，存放到 public/uploads/20261028/{md5}.png 2. 写入本地 fa_attachment 附件表与 fa_fc_category 分类表 3. 将新增 uploads 文件及静态一级图标 SFTP 同步至线上服务器 4. 将本地 fa_fc_category 与 fa_attachment 变更同步至线上数据库 5. 清除缓存并校验线上接口 """ import os import sys import io import time import hashlib import paramiko import pymysql sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') # 本地路径 LOCAL_ROOT = r"D:\Softwares\phpstudy_pro\WWW\fenleixinxi" PUBLIC_DIR = os.path.join(LOCAL_ROOT, "public") STATIC_CAT_DIR = os.path.join(PUBLIC_DIR, "static", "categories") DATE_DIR = "20261028" UPLOADS_TARGET_DIR = os.path.join(PUBLIC_DIR, "uploads", DATE_DIR) # 远程配置 REMOTE_HOST = "8.137.181.63" REMOTE_PORT = 22 REMOTE_USER = "root" REMOTE_PASS = "liyanxia@6688" REMOTE_ROOT = "/www/wwwroot/fenleixinxi.gzyakj.com.cn" DB_USER = "fenleixinxi" DB_PASS = "fenleixinxi" DB_NAME = "fenleixinxi" # 本地数据库配置 LOCAL_DB_HOST = "127.0.0.1" LOCAL_DB_PORT = 3307 # 13个二级分类映射关系: (分类ID, 本地原文件名, 分类名) SUB_CATS = [ (27, "job-tech.png", "普工技工"), (28, "job-catering.png", "餐饮服务"), (29, "job-clean.png", "家政保洁"), (30, "job-rider.png", "骑手配送"), (31, "job-sales.png", "销售客服"), (32, "job-driver.png", "司机出行"), (33, "job-edu.png", "教育培训"), (34, "job-parttime.png", "兼职短工"), (35, "poultry-egg.png", "鲜土鸡蛋"), (36, "poultry-chicken.png", "散养土鸡"), (37, "poultry-pork.png", "生鲜猪肉"), (38, "poultry-beef.png", "高山牛肉"), (39, "poultry-fish.png", "鲜活水产"), ] def ssh_exec(client, cmd): stdin, stdout, stderr = client.exec_command(cmd) out = stdout.read().decode("utf-8", errors="replace").strip() err = stderr.read().decode("utf-8", errors="replace").strip() return out, err def main(): print("=== [第一步] 模拟真实上传：将二级分类图标移入 uploads 目录 ===") os.makedirs(UPLOADS_TARGET_DIR, exist_ok=True) # 连接本地数据库 local_conn = pymysql.connect( host=LOCAL_DB_HOST, port=LOCAL_DB_PORT, user=DB_USER, password=DB_PASS, database=DB_NAME, charset="utf8mb4", autocommit=True ) uploaded_files = [] # 保存 (本地绝对路径, 相对远程路径, URL) with local_conn.cursor() as cur: for cat_id, filename, cat_name in SUB_CATS: src_path = os.path.join(STATIC_CAT_DIR, filename) if not os.path.exists(src_path): print(f" ✗ 未找到原始图标文件: {src_path}") continue with open(src_path, "rb") as f: content = f.read() file_md5 = hashlib.md5(content).hexdigest() file_sha1 = hashlib.sha1(content).hexdigest() file_size = len(content) dest_filename = f"{file_md5}.png" dest_abs_path = os.path.join(UPLOADS_TARGET_DIR, dest_filename) # 写入本地 uploads with open(dest_abs_path, "wb") as f: f.write(content) url_path = f"/uploads/{DATE_DIR}/{dest_filename}" uploaded_files.append((dest_abs_path, f"public/uploads/{DATE_DIR}/{dest_filename}", url_path)) now_ts = int(time.time()) # 1. 写入/更新本地 fa_attachment cur.execute("SELECT id FROM fa_attachment WHERE url = %s LIMIT 1", (url_path,)) att = cur.fetchone() if not att: cur.execute(""" INSERT INTO fa_attachment (category, admin_id, user_id, url, imagewidth, imageheight, imagetype, imageframes, filename, filesize, mimetype, createtime, updatetime, uploadtime, storage, sha1) VALUES ('category', 1, 0, %s, 240, 240, 'png', 0, %s, %s, 'image/png', %s, %s, %s, 'local', %s) """, (url_path, filename, file_size, now_ts, now_ts, now_ts, file_sha1)) # 2. 更新本地 fa_fc_category 记录的 image 字段 cur.execute("UPDATE fa_fc_category SET image = %s, updatetime = %s WHERE id = %s", (url_path, now_ts, cat_id)) print(f" ✓ 分类 [{cat_name}] (ID:{cat_id}) -> 模拟上传为: {url_path}") print("\n=== [第二步] 连接线上服务器并上传文件 (SFTP) ===") ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(REMOTE_HOST, port=REMOTE_PORT, username=REMOTE_USER, password=REMOTE_PASS, timeout=30) print("✓ 远程 SSH 连接成功") # 创建远程 uploads 目录 remote_upload_dir = f"{REMOTE_ROOT}/public/uploads/{DATE_DIR}" ssh_exec(ssh, f"mkdir -p {remote_upload_dir}") ssh_exec(ssh, f"mkdir -p {REMOTE_ROOT}/public/static/categories") sftp = ssh.open_sftp() # 1. 上传新生成的二级分类 uploads 图片 print(f"正在上传 {len(uploaded_files)} 个二级分类图标到线上 uploads 目录...") for local_file, remote_rel_file, url in uploaded_files: remote_full = f"{REMOTE_ROOT}/{remote_rel_file}" sftp.put(local_file, remote_full) ssh_exec(ssh, f"chown www:www {remote_full}") print("✓ 二级分类图片已全量上传至线上 uploads 目录") # 2. 上传一级分类图标 (cat-job.png, cat-poultry.png) parent_icons = ["cat-job.png", "cat-poultry.png"] for icon_name in parent_icons: loc = os.path.join(STATIC_CAT_DIR, icon_name) if os.path.exists(loc): rem = f"{REMOTE_ROOT}/public/static/categories/{icon_name}" sftp.put(loc, rem) ssh_exec(ssh, f"chown www:www {rem}") print(f" ✓ 已同步一级图标: {icon_name}") sftp.close() print("\n=== [第三步] 同步本地分类数据与附件记录至线上数据库 ===") # 导出本地分类表 SQL with local_conn.cursor() as cur: cur.execute("SELECT * FROM fa_fc_category;") cat_rows = cur.fetchall() cur.execute("DESCRIBE fa_fc_category;") cat_cols = [f[0] for f in cur.fetchall()] # 查询新增的 attachment 记录 cur.execute(f"SELECT * FROM fa_attachment WHERE url LIKE '/uploads/{DATE_DIR}/%';") att_rows = cur.fetchall() cur.execute("DESCRIBE fa_attachment;") att_cols = [f[0] for f in cur.fetchall()] # 构造同步 SQL sync_sqls = ["SET FOREIGN_KEY_CHECKS = 0;"] # 1. 线上清空并全量重写 fa_fc_category，确保和本地 100% 对齐 sync_sqls.append("TRUNCATE TABLE fa_fc_category;") col_str = ", ".join([f"`{c}`" for c in cat_cols]) for row in cat_rows: val_list = [] for v in row: if v is None: val_list.append("NULL") elif isinstance(v, (int, float)): val_list.append(str(v)) else: esc = str(v).replace("\\", "\\\\").replace("'", "\\'") val_list.append(f"'{esc}'") sync_sqls.append(f"INSERT INTO fa_fc_category ({col_str}) VALUES ({', '.join(val_list)});") # 2. 插入 attachment 记录 if att_rows: att_col_str = ", ".join([f"`{c}`" for c in att_cols]) for row in att_rows: val_list = [] for v in row: if v is None: val_list.append("NULL") elif isinstance(v, (int, float)): val_list.append(str(v)) else: esc = str(v).replace("\\", "\\\\").replace("'", "\\'") val_list.append(f"'{esc}'") sync_sqls.append(f"INSERT IGNORE INTO fa_attachment ({att_col_str}) VALUES ({', '.join(val_list)});") sync_sqls.append("SET FOREIGN_KEY_CHECKS = 1;") full_sql = "\n".join(sync_sqls) # 写入远程执行 mysql_cmd = f'mysql -u{DB_USER} -p{DB_PASS} --default-character-set=utf8mb4 {DB_NAME} 2>/dev/null <<\'SQLEOF\'\n{full_sql}\nSQLEOF' out, err = ssh_exec(ssh, mysql_cmd) if err and "error" in err.lower(): print(f"✗ 线上数据库更新失败: {err}") else: print("✓ 线上数据库 fa_fc_category 与 fa_attachment 已经完美同步！") print("\n=== [第四步] 清理缓存与验证 ===") ssh_exec(ssh, f"rm -rf {REMOTE_ROOT}/runtime/temp/* {REMOTE_ROOT}/runtime/cache/*") print("✓ 线上运行时缓存清理完成") # 校验线上分类数量 out, _ = ssh_exec(ssh, f'mysql -u{DB_USER} -p{DB_PASS} {DB_NAME} -e "SELECT COUNT(*) FROM fa_fc_category;" 2>/dev/null') print(f" 线上分类总数: {out.strip().split(chr(10))[-1].strip()} 条 (本地: {len(cat_rows)} 条)") # 测试线上图片 URL 可访问性 sample_url = uploaded_files[0][2] test_img_url = f"https://fenleixinxi.gzyakj.com.cn{sample_url}" cmd_curl = f"curl -s -o /dev/null -w '%{{http_code}}' -k '{test_img_url}'" code, _ = ssh_exec(ssh, cmd_curl) print(f" 线上二级分类上传图片可访问性验证 ({sample_url}): HTTP {code}") # 测试线上分类 API api_url = "https://fenleixinxi.gzyakj.com.cn/api/v1/home/categories" cmd_api = f"curl -s -k '{api_url}'" api_res, _ = ssh_exec(ssh, cmd_api) if '"code":1' in api_res and sample_url in api_res: print(" ✓ 线上分类 API 成功返回新二级分类及真实 uploads 图片地址！") else: print(f" API 响应采样: {api_res[:200]}") ssh.close() local_conn.close() print("\n✓ 全部同步工作圆满完成！") if __name__ == "__main__": main()