不用下载记账 APP!WorkBuddy+Python,聊天就能记账

import sqlite3import csvimport jsonfrom datetime import datetimeclass AccountBook:    def __init__(self, db_name="account.db"):        self.conn = sqlite3.connect(db_name, check_same_thread=False)        self.cursor = self.conn.cursor()        self.create_table()    def create_table(self):        """创建记账表"""        sql = '''        CREATE TABLE IF NOT EXISTS bill (            id INTEGER PRIMARY KEY AUTOINCREMENT,            bill_type TEXT NOT NULL, -- income收入 / expense支出            category TEXT NOT NULL, --分类：餐饮、交通、购物、工资等            amount REAL NOT NULL, --金额            remark TEXT, --备注            create_time TEXT NOT NULL --记录时间 yyyy‑mm‑dd        )        '''        self.cursor.execute(sql)        self.conn.commit()    def add_bill(self, bill_type, category, amount, remark="", create_time=None):        """新增账单        bill_type: income 收入 / expense 支出        """        if create_time is None:            create_time = datetime.now().strftime("%Y-%m-%d")        try:            self.cursor.execute(                "INSERT INTO bill(bill_type,category,amount,remark,create_time) VALUES (?,?,?,?,?)",                (bill_type, category, amount, remark, create_time)            )            self.conn.commit()            return {"code":0, "msg":"记账成功"}        except Exception as e:            return {"code":-1, "msg":f"记账失败:{str(e)}"}    def query_all(self):        """查询全部账单"""        self.cursor.execute("SELECT * FROM bill ORDER BY create_time DESC")        res = self.cursor.fetchall()        cols = [c[0] for c in self.cursor.description]        data = [dict(zip(cols, row)) for row in res]        return {"code":0, "data":data}    def query_by_date(self, start_date, end_date):        """按时间范围查询账单"""        self.cursor.execute(            "SELECT * FROM bill WHERE create_time BETWEEN ? AND ? ORDER BY create_time DESC",            (start_date, end_date)        )        res = self.cursor.fetchall()        cols = [c[0] for c in self.cursor.description]        data = [dict(zip(cols, row)) for row in res]        return {"code":0, "data":data}    def delete_bill(self, bill_id):        """根据id删除账单"""        self.cursor.execute("DELETE FROM bill WHERE id=?", (bill_id,))        self.conn.commit()        if self.cursor.rowcount == 0:            return {"code":-1, "msg":"未找到该账单"}        return {"code":0, "msg":"删除成功"}    def update_bill(self, bill_id, bill_type=None, category=None, amount=None, remark=None):        """修改账单记录"""        fields = []        params = []        if bill_type:            fields.append("bill_type=?")            params.append(bill_type)        if category:            fields.append("category=?")            params.append(category)        if amount is not None:            fields.append("amount=?")            params.append(amount)        if remark is not None:            fields.append("remark=?")            params.append(remark)        params.append(bill_id)        if not fields:            return {"code":-1, "msg":"没有要修改的字段"}        sql = f"UPDATE bill SET {','.join(fields)} WHERE id=?"        self.cursor.execute(sql, params)        self.conn.commit()        return {"code":0, "msg":"修改完成"}    def get_stat(self, start_date=None, end_date=None):        """收支统计：总收入、总支出、结余、分类汇总"""        if start_date and end_date:            cond = "WHERE create_time BETWEEN ? AND ?"            args = (start_date, end_date)        else:            cond = ""            args = ()        # 总收入、总支出        self.cursor.execute(f'''        SELECT        SUM(CASE WHEN bill_type='income' THEN amount ELSE 0 END) as total_income,        SUM(CASE WHEN bill_type='expense' THEN amount ELSE 0 END) as total_expense        FROM bill {cond}        ''', args)        row = self.cursor.fetchone()        total_income = row[0] if row[0] else 0        total_expense = row[1] if row[1] else 0        balance = total_income - total_expense        # 分类支出统计        self.cursor.execute(f'''        SELECT category, SUM(amount) as sum_amount        FROM bill WHERE bill_type='expense' {cond}        GROUP BY category ORDER BY sum_amount DESC        ''', args)        cat_data = [{"category":r[0], "sum_amount":round(r[1],2)} for r in self.cursor.fetchall()]        return {            "code":0,            "data":{                "total_income": round(total_income,2),                "total_expense": round(total_expense,2),                "balance": round(balance,2),                "category_expense": cat_data            }        }    def export_csv(self, filename="bill_out.csv"):        """导出账单csv"""        ret = self.query_all()        if ret["code"] !=0 or len(ret["data"])==0:            return {"code":-1,"msg":"暂无账单数据，无法导出"}        with open(filename,"w",newline="",encoding="utf‑8‑sig") as f:            writer = csv.DictWriter(f, fieldnames=ret["data"][0].keys())            writer.writeheader()            writer.writerows(ret["data"])        return {"code":0,"msg":f"导出文件 {filename}"}    def close(self):        self.conn.close()# 本地控制台测试入口if __name__ == "__main__":    ab = AccountBook()    print("====小型记账工具====")    print("1新增账单 2全部账单 3时间段查询 4修改账单 5删除账单 6统计 7导出 0退出")    while True:        op = input("\n输入操作序号：")        if op == "0":            ab.close()            break        elif op == "1":            btype = input("类型(income收入/expense支出):")            cat = input("分类(餐饮/交通等):")            amt = float(input("金额:"))            rem = input("备注:")            print(ab.add_bill(btype,cat,amt,rem))        elif op == "2":            print(json.dumps(ab.query_all(),ensure_ascii=False,indent=2))        elif op == "3":            s = input("开始日期(yyyy‑mm‑dd):")            e = input("结束日期(yyyy‑mm‑dd):")            print(json.dumps(ab.query_by_date(s,e),ensure_ascii=False,indent=2))        elif op == "4":            bid = int(input("账单ID:"))            bt = input("类型(回车不修改):") or None            ct = input("分类(回车不修改):") or None            am_in = input("金额(回车不修改):")            am = float(am_in) if am_in else None            re = input("备注(回车不修改):") or None            print(ab.update_bill(bid, bt, ct, am, re))        elif op == "5":            bid = int(input("账单ID:"))            print(ab.delete_bill(bid))        elif op == "6":            print(json.dumps(ab.get_stat(),ensure_ascii=False,indent=2))        elif op == "7":            print(ab.export_csv())

pip install fastapi uvicorn

from fastapi import FastAPIfrom pydantic import BaseModelfrom account_book import AccountBookapp = FastAPI(title="WorkBuddy记账工具API")ab = AccountBook()class BillAdd(BaseModel):    bill_type: str    category: str    amount: float    remark: str = ""    create_time: str = Noneclass BillUpdate(BaseModel):    bill_id:int    bill_type:str=None    category:str=None    amount:float=None    remark:str=None@app.post("/bill/add")def add(bill:BillAdd):    return ab.add_bill(bill.bill_type,bill.category,bill.amount,bill.remark,bill.create_time)@app.get("/bill/all")def all_bill():    return ab.query_all()@app.get("/bill/range")def range_bill(start_date:str, end_date:str):    return ab.query_by_date(start_date, end_date)@app.post("/bill/update")def update(bill:BillUpdate):    return ab.update_bill(bill.bill_id, bill.bill_type, bill.category, bill.amount, bill.remark)@app.delete("/bill/del/{bill_id}")def delete(bill_id:int):    return ab.delete_bill(bill_id)@app.get("/bill/stat")def stat(start_date:str=None, end_date:str=None):    return ab.get_stat(start_date, end_date)@app.get("/bill/export")def export():    return ab.export_csv()if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="0.0.0.0", port=8001)

你是AI记账助手，用户用自然语言记账，调用插件完成记账。支持：记收入、记支出、查询账单、时间段查账单、修改、删除账单、收支统计、导出账单。示例指令：记一笔支出，分类餐饮，金额35元，备注晚饭记一笔收入，工资，8000元查看本月账单统计本月收支情况删除ID为3的账单导出全部账单返回结果整理成清晰表格给用户。