77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""导入王芳邮储卡外部真实消费（排除支付宝/微信渠道重复、内部自转、家庭内部）"""
|
|
import csv, re, json, pymysql
|
|
from datetime import datetime
|
|
cfg = json.load(open('db_config.json'))
|
|
conn = pymysql.connect(host=cfg['host'], port=cfg['port'], user=cfg['user'],
|
|
password=cfg.get('password'), database=cfg['database'], charset='utf8mb4')
|
|
cur = conn.cursor()
|
|
MEMBER = 3
|
|
|
|
rows = []
|
|
with open('/tmp/psbc_records.csv', encoding='utf-8') as f:
|
|
for row in csv.reader(f):
|
|
if not row or len(row) < 3:
|
|
continue
|
|
try:
|
|
float(row[1])
|
|
except:
|
|
continue
|
|
rows.append(row)
|
|
|
|
PROP_CARDS = {'621775831', '621778830', '622203130', '567156987', '208860212'}
|
|
alipay_pat = re.compile('2088\\d{11}')
|
|
|
|
def is_external_expense(r):
|
|
amt = float(r[1]); s = r[3]
|
|
if amt >= 0: return False
|
|
if '退货' in s: return False
|
|
if alipay_pat.search(s): return False
|
|
if '财付通' in s or '微信' in s: return False
|
|
for c in PROP_CARDS:
|
|
if c in s: return False
|
|
if '杨轩' in s or '汪礼平' in s or '汪和平' in s: return False
|
|
if '还款' in s: return False
|
|
return True
|
|
|
|
def cat_of(s):
|
|
if any(k in s for k in ['超市','生鲜','便利店','水果','餐饮','烧烤','龙虾','包子','蜜雪','瑞幸','金拱门','餐厅','食堂','食品','桃酥','米糕','药材','调料']):
|
|
return '餐饮/食品'
|
|
if any(k in s for k in ['铁路','12306','滴滴','停车','地铁','交通','火车','顺风','轨道']):
|
|
return '交通'
|
|
if any(k in s for k in ['药','医院','诊所','健康']):
|
|
return '医疗'
|
|
if any(k in s for k in ['母婴','奶粉','尿不湿']):
|
|
return '母婴'
|
|
if any(k in s for k in ['童装','服装','女装','百货','鞋','服饰']):
|
|
return '购物'
|
|
if any(k in s for k in ['美发','美容']):
|
|
return '美容美发'
|
|
if any(k in s for k in ['缴费','水电','话费','燃气']):
|
|
return '生活缴费'
|
|
if any(k in s for k in ['游泳','健身','运动']):
|
|
return '运动'
|
|
if any(k in s for k in ['快团团','拼多多']):
|
|
return '购物'
|
|
if any(k in s for k in ['美团','饿了么','拉扎斯','三快','外卖']):
|
|
return '餐饮/食品'
|
|
return '其他'
|
|
|
|
ext_count = 0; ext_sum = 0.0; cat_tot = {}
|
|
for r in rows:
|
|
if is_external_expense(r):
|
|
amt = abs(float(r[1])); s = r[3]
|
|
dt = datetime.strptime(r[0], '%Y-%m-%d %H:%M:%S')
|
|
c = cat_of(s)
|
|
note = f"[邮储卡] 外部消费|{s.strip()}"
|
|
cur.execute("INSERT INTO variable_expenses (member_id,category,amount,recorded_date,note) VALUES (%s,%s,%s,%s,%s)",
|
|
(MEMBER, c, amt, dt.date(), note))
|
|
ext_count += 1; ext_sum += amt; cat_tot[c] = cat_tot.get(c, 0) + amt
|
|
|
|
conn.commit()
|
|
print(f"外部消费导入: {ext_count}笔 ¥{ext_sum:.2f}")
|
|
for k, v in sorted(cat_tot.items(), key=lambda x: -x[1]):
|
|
print(f" {k:10s} ¥{v:.2f}")
|
|
conn.close()
|