简单介绍

因为本人平常有记账的习惯，所以突发奇想，想做一个可以记账的公众号。
先贴上公众号的二维码，有兴趣的朋友可以尝试一下。

具体使用方法：
1、消费记录
可以把一些日常消费按照一个格式发送给公众号，格式为"时间消费项目金额"或"消费项目金额"，时间和消费项目、消费项目和消费金额之间一定要加空格，例如"2.17 早餐 6"、“午餐 15”，时间格式目前只支持"月份.日期" ，如果不加时间，则默认记录在当天的消费中；消费金额以元为单位。发送消费记录到公众号之后，公众号就会帮你自己动记录。也可以一次发送多个消费记录，以换行分隔，例如下图

2、统计消费金额
发送"统计"到公众后，公众号就会返回当月消费、当天消费、当月日均消费的情况，具体见上图。
3、将消费记录生成csv文件、日均消费趋势图
发送"表格"到公众后，公众号会返回一个链接，打开链接可以下载一个包含当月所有消费的csv文件，见下图。
发送"图表"到公众后，公众号会返回一个链接，打开链接可以看到当月的一个日均消费趋势情况，见下图。

4、撤销
发送"撤销"到公众号，可以撤销发送的上一条消费记录。*

实现方法

整个服务是通过 flask 来实现的，就是在微信公众号绑定的ip服务器上开启一个 flask 服务，来接收用户发来的信息，检查信息中是否包含之前预设好的一些关键词，来调用相应的函数实现服务。
用到的第三方库：flask、wechatpy、pymysql、DBUtils、pyecharts。
个人感觉有些逻辑实现的还是太复杂了，欢迎朋友们指出不足的地方，谢谢！

完整代码请移步：https://github.com/a596480606/wechat_public

main.py

from flask import Flask, request
from wechatpy import parse_message
from table_funs import get_day_data, gen_day_table
from wechatpy.replies import TextReply
from tools import parse_items, get_today_date, get_consume, get_avg_consume, gen_table, Mymysql, RegexConverterserver_ip = "your server ip"# 创建 flask 应用
app = Flask(__name__)
app.url_map.converters['re'] = RegexConverter   # 通过正则匹配路由# 暂时存放用户对象的字典，后期可替换成redis
USERS = {}# 实例化数据库对象
DB = Mymysql()@app.route("/<re('.?'):url>", methods=["GET", "POST"])
def index(url):global USERSprint("收到请求！",url)# 解析收到的用户数据data = parse_message(request.data)print("data:", data)if not data:return "无效的请求！"# 提取出用户发送给公众号的信息和用户名msg = data.contentuser = data.source.replace("@", "_").replace("-","_")# 如果用户不在USERS中，则为其新建一张表if user not in USERS:USERS[user] = {}DB.create(user)# 用户发送的信息中包含统计if "统计" in msg:today = get_today_date()            # 获取当天日期all_consume = DB.select(user)       # 获取用户数据中的所有记录# 统计出日消费和月总消费day_consume, month_consume = get_consume(all_consume,  today)avg_consume = get_avg_consume(month_consume)reply_text = "今日消费：" + str(day_consume) + "元\n当月消费：" + str(month_consume) + "元\n日均消费：" + str(avg_consume) + "元"return TextReply(content=reply_text, message=data).render()# 用户发送的信息中包含表格elif "表格" in msg:all_consume = DB.select(user)          # 获取数据库中关于用户的所有记录path = f"http://{server_ip}:8015/" + gen_table(user, all_consume)reply_text = pathreturn TextReply(content=reply_text, message=data).render()# 用户发送的信息中包含撤销elif "撤销" in msg:res = DB.get_last_record(user)          # 获取数据库中关于用户的最后一条记录if not res:return TextReply(content="暂无数据可撤销！", message=data).render()last_record = eval(res)reply_text = f"已撤销  {' '.join(last_record)} 的记录！"DB.drop_a_record(user)return TextReply(content=reply_text, message=data).render()# 用户发送的信息中包含图表elif '图表' in msg:dates, values = get_day_data(user, DB)gen_day_table(user,dates,values)url = f"http://{server_ip}:8015/table/{user}"return TextReply(content=url, message=data).render()# 切割提取用户发送的消费信息try:items = [item for item in msg.split("\n")]items = [[xs for xs in s.split(" ") if xs] for s in items if s]except:print("split error")reply_text = "数据格式不正确！"return TextReply(content=reply_text, message=data).render()# 向数据库插入记录成功的数据，并返回记录失败的数据results = "以下数据记录失败:\n"for num, item in enumerate(items):res = parse_items(item, user, DB)print(res)if res != "已记录!":results += "  ".join(item) + "\n"if results == "以下数据记录失败:\n":reply_text = "记录成功！"else:reply_text = resultsxml = TextReply(content=reply_text, message=data).render()return xmlif __name__ == '__main__':app.run(host="0.0.0.0", port=80)

主要的工具函数
tools.py

"""main.py 用到的函数集合
"""
import csv
import time
import pymysql
import datetime
from DBUtils.PooledDB import PooledDB
from werkzeug.routing import BaseConverterclass RegexConverter(BaseConverter):def __init__(self, url_map, *args):super(RegexConverter, self).__init__(url_map)# 将接受的第1个参数当作匹配规则进行保存self.regex = args[0]def parse_items(item, user, DB):"""处理用户消费信息的函数:param item: 消费记录:param user: 用户名:param DB:  数据库:return:"""if len(item) == 2:today = get_today_date()if item[0].split(".")[0].isdigit():reply_text = "格式错误！请输入例如：早餐 12"return reply_textif is_number(item[1]):item.insert(0, today)DB.insert(user,str(item))reply_text = "已记录!"else:reply_text = "格式错误！请输入例如：早餐 12"elif len(item) == 3:today = format_date(item[0])if not today:reply_text = "日期格式不正确！请输入例如：12.2 或 12.02"return reply_textif is_number(item[2]):reply_text = "已记录!"item[0] = todayDB.insert(user,str(item))else:reply_text = "格式错误！请输入例如：早餐 12"else:reply_text = "暂不支持其他消息，请见谅！"return reply_textdef get_today_date(detail=False):"""detail 日期细节，默认False，返回到日, 为 s 返回 到时分秒"""if not detail:t = time.strftime("%Y%m%d", time.localtime())return tif detail == "s":t = time.strftime("%Y%m%d%H%M%S", time.localtime())return tdef time_to_timestmap(t):"""把 类似 20191122 格式的时间转为时间戳"""new_t = time.mktime(time.strptime(t, "%Y%m%d"))return round(new_t)def get_current_month():"""返回当前月份的第一天的日期 20191201"""year = datetime.datetime.today().yearmonth = datetime.datetime.today().monthtomonth = str(year) + str(month) + "01"return tomonthdef get_consume(all_consume, today):"""统计用户消费的函数:param all_consume: 数据库中关于用户的所有记录:param today: 当天日期:return: 日总消费，月总消费"""tomonth = get_current_month()       # 当月第一天的日期month_consume = 0                   # 月总消费today_consume = 0                   # 日总消费# 遍历数据库中关于用户的所有记录for consume in all_consume:record = eval(consume[0])consume_day = record[0]fee = float(record[2])if consume_day == today:today_consume += feeif time_to_timestmap(consume_day) >= time_to_timestmap(tomonth):month_consume += float(fee)return round(today_consume,2), round(month_consume,2)def get_avg_consume(month_consume):"""生成月平均消费的函数"""day = datetime.datetime.today().dayavg_consume = round(month_consume/day,2)return avg_consumedef format_date(s):year = str(datetime.datetime.today().year)if "." in s:sl = s.split(".")else:return Falseif len(sl) != 2:return Falsemonth, day = slif month.isdigit() and day.isdigit():month, day = int(month),  int(day)else:return Falseif 0 < month < 10:month = "0" + str(month)elif 10 <= month < 13:month = str(month)else:return Falseevery_month_last_day = 31if int(month) in [1,3,5,7,8,10,12]:every_month_last_day = 32if 0 < day < 10:day = "0" + str(day)elif 10 <= day < every_month_last_day:day = str(day)else:return Falsereturn year + month + daydef is_number(string):try:float(string)except ValueError:return Falsereturn stringdef gen_table(user, all_consume):"""生产用户消费表格的函数:param user:    用户名:param all_consume: 获取数据库中关于用户的所有记录:return: 用户消费表格的路径"""path = f"{user}_details.csv"with open(f"./forms/{user}_details.csv", "w", encoding="utf8", newline="" ) as f:csvf = csv.writer(f, delimiter=",")for consume in all_consume:record = eval(consume[0])csvf.writerow(record)return pathclass Mymysql:"""mysql数据库类"""def __init__(self):# 通过 PooledDB 生成一个连接数最大为10的连接池self.pool = PooledDB(pymysql, 10, host="127.0.0.1", port=3306, user="root", password="123456", db="accounts")def create(self, user):self.conn = self.pool.connection()self.cursor = self.conn.cursor()# self.conn.cursor.execute("""DROP TABLE IF EXISTS {};""".format(user))sql = """create table IF NOT EXISTS {}( content varchar(50), id INT(20) AUTO_INCREMENT, primary key(id));""".format(user)print(sql)self.conn.ping(reconnect=True)self.cursor.execute(sql)self.cursor.close()self.conn.close()def insert(self, user, value):self.conn = self.pool.connection()self.cursor = self.conn.cursor()sql = """insert into {}(content) value ("{}");""".format(user,value)print(sql)self.conn.ping(reconnect=True)self.cursor.execute(sql)self.conn.commit()self.cursor.close()self.conn.close()def get_last_record(self, user):self.conn = self.pool.connection()self.cursor = self.conn.cursor()sql = """select * from {} order by id desc limit 1;""".format(user)print(sql)self.conn.ping(reconnect=True)self.cursor.execute(sql)try:res = self.cursor.fetchall()[0][0]except IndexError:return Noneprint(type(res))print(res)self.cursor.close()self.conn.close()return resdef drop_a_record(self, user):self.conn = self.pool.connection()self.cursor = self.conn.cursor()# sql = """delete from {} where id = (select id from {} Limit (count-1),1);""".format(user,user)sql = """delete from {}  order by id desc limit 1;""".format(user)print(sql)self.conn.ping(reconnect=True)self.cursor.execute(sql)self.conn.commit()self.cursor.close()self.conn.close()def select(self,user):self.conn = self.pool.connection()self.cursor = self.conn.cursor()sql = """select * from {};""".format(user)print(sql)self.conn.ping(reconnect=True)self.cursor.execute(sql)try:res = self.cursor.fetchall()except IndexError:return Noneself.cursor.close()self.conn.close()return res

用python开发一个记账的微信公众号相关推荐

python爬虫实战-爬取微信公众号所有历史文章 - (00) 概述
http://efonfighting.imwork.net 欢迎关注微信公众号"一番码客"获取免费下载服务与源码,并及时接收最新文章推送. 最近几年随着人工智能和大数据的兴起,p ...
python3怎么创建一个链表_怎么创建一个自己的微信公众号
随着移动互联网的发展,微信公众号这两年成为媒体传播的重要平台.平时我们每个人都会接触见到到各种非常好的微信公众号内容,碰到喜欢的有帮助的也会自觉的转发朋友圈或者群,分享给自己身边的人.于是很多人就想了 ...
微信公众号Java开发-笔记01【微信公众号介绍、开发环境搭建】
学习网址:哔哩哔哩网站微信公众号开发-Java版微信公众号Java开发-笔记01[微信公众号介绍.开发环境搭建] 微信公众号Java开发-笔记02[] 微信公众号Java开发-笔记03[] 微信公 ...
python wechatsougou_python抓取搜狗微信公众号文章
初学python,抓取搜狗微信公众号文章存入mysql mysql表: 代码: import requests import json import re import pymysql # 创建连接 ...
微信公众号开发用书php,php微信公众号开发（3）php实现简单微信文本通讯
<PHP实战:PHP微信公众号开发(3)PHP实现简单微信文本通讯>要点: 本文介绍了PHP实战:PHP微信公众号开发(3)PHP实现简单微信文本通讯,希望对您有用.如果有疑问,可以联系我 ...
python通过手机抓取微信公众号
使用 Fiddler 抓包分析公众号打开微信随便选择一个公众号,查看公众号的所有历史文章列表在 Fiddler 上已经能看到有请求进来了,说明公众号的文章走的都是HTTPS协议,这些请求就是微信客 ...
url 微信公众号开发配置失效_微信公众号开发之授权登录
一.UnionId和openId 微信登录最重要的两个返回信息,一个是UnionId,一个是OpenId.两者之间有着必然的联系. UnionID机制的作用说明:如果开发者拥有多个移动应用.网站应用和 ...
[python]用flask框架搭建微信公众号的后台
用flask框架搭建微信公众号的后台最近用python写了点爬虫,为了要让爬取的数据能够随时显示在我眼前,并实时根据我的指令返回数据.于是采用微信公众号做这个显示窗口,既能发送指令也能显示简单的相关 ...
php公众号开发配置网页域名,微信公众号网页开发授权配置流程
1.配置 1.1 公众号配置 1.2 开发者账号配置选择开发者工具->web开发者工具->绑定开发者微信账号. 1.3 开发域名配置选择公众号设置->功能设置,根据需求设置:业务 ...

用python开发一个记账的微信公众号

简单介绍

实现方法

用python开发一个记账的微信公众号相关推荐

最新文章

热门文章