GuiZhouChengXuRuanJianKaiFaLiYanXia/fenleixinxi
1
0
派生 0
代码 工单 合并请求 Actions 软件包 项目 版本发布 百科 动态
文件
f13fb6cdab064c9a2dbf7b7f40dced519995d31f
fenleixinxi/tools/simulate_upload_and_sync.py
MiAn f13fb6cdab feat: 完成二级分类真实实拍图替换与功能优化 
本次提交内容包括：
1. 替换13个二级分类为真实商业实拍图，同步更新本地与线上数据库
2. 新增会员拨打额度耗尽引导开通会员功能
3. 优化发布页分类选择逻辑，增加会员权益展示
4. 修复分类名称显示问题，完善相关交互逻辑
5. 新增图片同步工具脚本，确保线上线下数据一致性
2026-10-28 14:13:32 +08:00

244 行
9.6 KiB
Python
原始文件 Blame 文件历史

#!/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()
在新工单中引用 查看 Git Blame 复制永久链接