手机网站的价值,content timeline wordpress,wordpress移动端顶部导航栏,东莞网站关键词优化公司介绍
采用微信的hook插件#xff0c;然后解析微信发来的数据图片#xff0c;通过ocr识别 然后将数据落入execl表格中。同时有权限的人可以导出数据表格即可。 流程图 代码片
文本消息处理流程_robot.py elif msg.type 0x01: # 文本消息# 管理员列表dba_user_list [wxid_…介绍
采用微信的hook插件然后解析微信发来的数据图片通过ocr识别 然后将数据落入execl表格中。同时有权限的人可以导出数据表格即可。 流程图 代码片
文本消息处理流程_robot.py
elif msg.type 0x01: # 文本消息# 管理员列表dba_user_list [wxid_uev4klri3snh22,zhangzijian0715,yanheng1984,wxid_30173uk0ohjd21]# 8812131740734078818 id字段# wxid_uev4klri3snh22 sender字段 我的WXIDself.LOG.info(发送人的ID: msg.sender ,发送内容: msg.content)# 管理员的特殊权限if msg.sender in dba_user_list:if msg.content.startswith(数据文件):# 发送文件self.wcf.send_file(data_execl_path,msg.sender)elif msg.content.startswith(添加人员):# 保存数据的字典data_dict {}data_list msg.content.split()if len(data_list) 4:data_dict[user_name] data_list[1]data_dict[user_department] data_list[2]data_dict[user_wxid] data_list[3]insert_data_result, insert_cause_str insert_user_data_info(data_dict)if insert_data_result False:insert_fail_info (数据插入失败请联系管理员处理cause: insert_cause_str)self.sendTextMsg(insert_fail_info, msg.sender)returnelse:insert_success_info (数据插入成功)self.sendTextMsg(insert_success_info, msg.sender)returnelse:self.sendTextMsg(数据格式错误,msg.sender)returnreturn# 让配置加载更灵活自己可以更新配置。也可以利用定时任务更新。# if msg.from_self():# if msg.content ^更新$:# self.config.reload()# self.LOG.info(已更新)# else:# self.toChitchat(msg) # 闲聊图片消息处理流程_robot.py
elif msg.type 3: # 图片消息try:##### 判断用户是否属于授权用户user_name get_user_name_info_by_wxid(msg.sender)if not user_name:self.sendTextMsg(你的账号暂未授权,请联系管理员授权账号\nwxid msg.sender, msg.sender)return# msg.extra字段为完整图片的dat路径,需要进行图片解密转换,另一个字段属于缩略图# DONE sleep是为了给图片预留落盘的时间不然会识别不到文件time.sleep(1)###### 下载Image目录的图片并返回图片地址current_datetime datetime.datetime.now()formatted_datetime current_datetime.strftime(%Y_%m).strip()# temp_pic目录的路径target_path os.path.join(os.path.join(os.getcwd(), temp_pic), formatted_datetime)if not os.path.exists(target_path):os.makedirs(target_path)self.wcf.download_image(msg.id, msg.extra, target_path)##### 根据图片地址OCR解析需要的内容## 标准精度识别# data_dict ocr_wx_pic(os.path.join(target_path, pic_name).replace(.dat, .jpg))## 高精度的ORC识别pic_name os.path.basename(msg.extra)new_pic_path os.path.join(target_path, pic_name).replace(.dat, .jpg)if not os.path.exists(new_pic_path):self.sendTextMsg(当前网络繁忙,图片解析失败,请稍后重试, msg.sender)self.LOG.error(当前网络繁忙,图片解析失败,请稍后重试,如有疑问请联系管理员)return### 加一层过滤 防止OCR识别发生错误或者异常图片不符合规范data_dict ocr_wx_pic_hign(new_pic_path)if data_dict None:self.sendTextMsg(图片不符合规范请上传本周小结的内容, msg.sender)self.LOG.error(图片不符合规范请上传本周小结的内容,如有疑问请联系管理员)return#### 判断数据是否已经录入数据库中如果未录入则录入data_is_exit get_user_week_info_is_exit(data_dict[week_date], user_name)data_dict[user_name] user_nameif data_is_exit False:insert_data_result,insert_cause_str insert_user_week_data_info(data_dict)if insert_data_result False:insert_fail_info (数据插入失败请联系管理员处理cause: insert_cause_str\n [数据总览]\n 姓名 : user_name \n 日期 : data_dict[week_date] \n 工作会话数 : str(data_dict[week_work_num]) 次\n 花费时长 : str(data_dict[week_work_total_min]) 分钟\n 最晚时间 : data_dict[week_final_last_time])self.sendTextMsg(insert_fail_info, msg.sender)returnelse:send_succ_info (数据插入成功\n [数据总览]\n 姓名 : user_name \n 日期 : data_dict[week_date] \n 工作会话数 : str(data_dict[week_work_num]) 次\n 花费时长 : str(data_dict[week_work_total_min]) 分钟\n 最晚时间 : data_dict[week_final_last_time])# 对内容进行转换self.sendTextMsg(send_succ_info, msg.sender)returnelse:# 对内容进行转换repeat_info (数据已存在,不可重复添加\n [数据总览]\n 姓名 : user_name \n 日期 : data_dict[week_date] \n 工作会话数 : str(data_dict[week_work_num]) 次\n 花费时长 : str(data_dict[week_work_total_min]) 分钟\n 最晚时间 : data_dict[week_final_last_time])self.sendTextMsg(repeat_info, msg.sender)returnexcept Exception as e:self.sendTextMsg(图片处理失败请联系管理员处理 cause str(e), msg.sender)self.LOG.exception(图片处理失败请联系管理员处理: %s, e)
execl表格的代码处理_csv_util.py
import pandas as pd
import os
import logging
import datetime# data_path os.path.join(os.path.dirname(os.path.dirname(os.getcwd())), data)data_path os.path.join(os.getcwd(), data)
data_execl_path os.path.join(data_path, weixin_data.xlsx)
user_execl_path os.path.join(data_path, user_info.xlsx)
logs logging.getLogger(csv_util)def read_excel(file_path):读取Excel文件并返回DataFrametry:os.chmod(file_path, 0o777)df pd.read_excel(file_path)return dfexcept Exception as e:logs.error(Error reading Excel file: {} , cause {},file_path, e)return Nonedef write_to_excel(file_path, new_row):将新数据写入Excel文件的最后一行try:df read_excel(file_path)if df is not None:# df df.append(data, ignore_indexTrue)df pd.concat([df, new_row.to_frame().T], ignore_indexTrue)# 将数据写入with pd.ExcelWriter(file_path, engineopenpyxl, modea, if_sheet_existsoverlay) as writer:df.to_excel(writer, indexFalse, sheet_nameSheet1)for key, value in new_row.items():logs.error(fKey: {key}, Value: {value})logs.info(Data written successfully. )return True,Noneelse:for key, value in new_row.items():logs.error(fKey: {key}, Value: {value})logs.warning(Failed to read Excel file., file info {} ,file_path)return False,Noneexcept Exception as e:for key, value in new_row.items():logs.error(fKey: {key}, Value: {value})logs.exception(Error writing to Excel file., file info {} ,cause info {},file_path, e)return False,str(e)def query_excel(file_path, field, value, return_fieldNone):根据指定字段查询数据try:df read_excel(file_path)if df is not None:result df[df[field] value]if return_field:return_values result[return_field]if not return_values.empty:return_values .join([return_values.iloc[0]])else:return_values .join(return_values)return str(return_values)return resultelse:logs.warning(Failed to read Excel file {} no found , field {} , value {} , return_field {},file_path, field, value, return_field)return Falseexcept Exception as e:logs.exception(Error querying Excel file {} , field {} , value {} , return_field {}, cause {},file_path, field, value, return_field, e)return False# 根据wxid获取用户的姓名
def get_user_name_info_by_wxid(wxid):return query_excel(user_execl_path, 微信唯一标识, wxid, 姓名)# 判断周报数据是否存在
def get_user_week_info_is_exit(data_str, user_name):result_df query_excel(data_execl_path, 日期, data_str)if result_df is None:return Falsefirst_column_values result_df[姓名].valuesif user_name in first_column_values:return Trueelse:return False# 将人员数据入库
def insert_user_data_info(data_dict):current_datetime datetime.datetime.now()formatted_datetime current_datetime.strftime(%Y-%m-%d %H:%M:%S).strip()new_data pd.Series({姓名: data_dict[user_name],部门: data_dict[user_department],微信唯一标识: data_dict[user_wxid],入库时间: formatted_datetime,# 添加更多列数据})# 将数据转换为DataFrame## new_row pd.Series(# [data_dict[user_name], data_dict[week_date], data_dict[week_work_num], data_dict[week_work_total_min],# data_dict[week_final_last_time], data_dict[data_year], formatted_datetime],# index[姓名, 日期, 工作会话数, 花费时长, 最晚时间, 年份, 入库时间])return write_to_excel(user_execl_path, new_data)# 将周报数据入库
def insert_user_week_data_info(data_dict):current_datetime datetime.datetime.now()formatted_datetime current_datetime.strftime(%Y-%m-%d %H:%M:%S).strip()new_data pd.Series({姓名: data_dict[user_name],日期: data_dict[week_date],工作会话数: data_dict[week_work_num],花费时长: data_dict[week_work_total_min],最晚时间: data_dict[week_final_last_time],年份: data_dict[data_year],入库时间: formatted_datetime,# 添加更多列数据})# 将数据转换为DataFrame## new_row pd.Series(# [data_dict[user_name], data_dict[week_date], data_dict[week_work_num], data_dict[week_work_total_min],# data_dict[week_final_last_time], data_dict[data_year], formatted_datetime],# index[姓名, 日期, 工作会话数, 花费时长, 最晚时间, 年份, 入库时间])return write_to_excel(data_execl_path, new_data)# 示例用法
if __name__ __main__:# 通过wxid 判断是否有权限# result get_user_name_info_by_wxid(wxid_uev4klri3snh22)result get_user_week_info_is_exit(20240603, 高垣)if result is not None:print(result)# file_path /mnt/data/excel_file.xlsx # 替换为你的Excel文件路径# excel_handler ExcelHandler(file_path)## # 读取Excel文件# df excel_handler.read_excel()# if df is not None:# print(df)## # 写入新数据# new_data {# Column1: Value1,# Column2: Value2,# # 添加更多列数据# }# excel_handler.write_to_excel(new_data)## # 根据指定字段查询数据# result excel_handler.query_excel(Column1, Value1)# if result is not None:# print(result)
图片ocr的图片处理_baidu_ocr.py
import base64
import urllib
import requests
import json
import re
import datetime
import os
import logging
import datetimeAPI_KEY XXXX
SECRET_KEY XXXXlogs logging.getLogger(baidu_ocr)
def main():url https://aip.baidubce.com/rest/2.0/ocr/v1/general?access_token get_access_token()# image 可以通过 get_file_content_as_base64(C:\fakepath\15866dbd4118eb7638c9a13b430dadf1.jpg,True) 方法获取# payload image%2F9j%2F4AAQSkZJRgABAQAAAQABAAD%2F2wBDAAEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQ...detect_directionfalsedetect_languagefalseparagraphfalseprobabilityfalsepayload imageget_file_content_as_base64(rE:\PythonCode\WeChatRobot\temp_pic\15866dbd4118eb7638c9a13b430dadf1.jpg,True)detect_directionfalsedetect_languagefalsevertexes_locationfalseparagraphfalseprobabilityfalseheaders {Content-Type: application/x-www-form-urlencoded,Accept: application/json}response requests.request(POST, url, headersheaders, datapayload)if(response.status_code200):pic_str json.loads(response.text)else:logs.error(接口请求失败。status_code {} , reason {},response.status_code,response.reason)def get_file_content_as_base64(path, urlencodedFalse):获取文件base64编码:param path: 文件路径:param urlencoded: 是否对结果进行urlencoded:return: base64编码信息with open(path, rb) as f:content base64.b64encode(f.read()).decode(utf8)if urlencoded:content urllib.parse.quote_plus(content)# print(content)return contentdef get_access_token():使用 AKSK 生成鉴权签名Access Token:return: access_token或是None(如果错误)url https://aip.baidubce.com/oauth/2.0/tokenparams {grant_type: client_credentials, client_id: API_KEY, client_secret: SECRET_KEY}return str(requests.post(url, paramsparams).json().get(access_token))def ocr_wx_pic(pic_full_path):url https://aip.baidubce.com/rest/2.0/ocr/v1/general?access_token get_access_token()payload image get_file_content_as_base64(pic_full_path,True) detect_directionfalsedetect_languagefalsevertexes_locationfalseparagraphfalseprobabilityfalseheaders {Content-Type: application/x-www-form-urlencoded,Accept: application/json}response requests.request(POST, url, headersheaders, datapayload)if response.status_code 200:#解析图片return parse_pic_data(response.text)else:print(接口请求失败。原因: response.reason)return None# 高精度版本
def ocr_wx_pic_hign(pic_full_path):url https://aip.baidubce.com/rest/2.0/ocr/v1/accurate_basic?access_token get_access_token()payload image get_file_content_as_base64(pic_full_path,True) detect_directionfalseparagraphfalseprobabilityfalseheaders {Content-Type: application/x-www-form-urlencoded,Accept: application/json}response requests.request(POST, url, headersheaders, datapayload)if response.status_code 200:#解析图片return parse_pic_data(response.text)else:logs.error(接口请求失败。status_code {} , reason {}, response.status_code, response.reason)return Nonedef parse_pic_data(orcStr):# 解析json字符串pic_str json.loads(orcStr)# 保存数据的字典data_dict {}# TODO 50表示该图片不是需要识别图片if pic_str[words_result_num] 50 or pic_str[words_result_num] 5:return None# 循环List去除掉无关项例如 手机图片头部信息 以及 尾部二维码等pic_list pic_str[words_result]# 循环pic_listfor idx,pic in enumerate(pic_list):# print(fpic: {pic})if pic[words].find(一周小结) ! -1:data_dict[week_date] pic[words].replace(一周小结, )elif pic[words].find(处理工作会话) ! -1:data_dict[week_work_num] extract_integer(pic_list[idx1][words])elif pic[words].find(总共花了) ! -1:data_dict[week_work_total_min] extract_integer(pic_list[idx1][words])elif pic[words].find(最晚时间) ! -1:data_dict[week_final_last_time] pic_list[idx1][words]# 手动添加年份data_dict[data_year] datetime.datetime.today().yearif len(data_dict) ! 5:return Nonereturn data_dictdef extract_integer(text):# 匹配字符串中的第一个整数部分match re.search(r\d, text)if match:# 找到整数则返回整数值return int(match.group())else:# 如果未找到整数则返回 None 或者其他你想要的默认值return Noneif __name__ __main__:pic_orc_str {words_result:[{words:中国移动1536●,location:{top:23,left:200,width:257,height:81}},{words:中国联通,location:{top:70,left:213,width:96,height:24}},{words:86令959l92乡,location:{top:46,left:667,width:370,height:44}},{words:一周小结5.27-5.31,location:{top:424,left:309,width:358,height:44}},{words:66,location:{top:573,left:242,width:49,height:37}},{words:金星虽然是不发光的行,location:{top:570,left:343,width:502,height:51}},{words:星但却比许多恒星还要,location:{top:640,left:343,width:555,height:51}},{words:明亮,location:{top:710,left:341,width:100,height:51}},{words:处理工作会话,location:{top:813,left:341,width:218,height:44}},{words:826次,location:{top:880,left:341,width:136,height:46}},{words:总共花了,location:{top:998,left:341,width:143,height:44}},{words:441分钟,location:{top:1066,left:341,width:177,height:46}},{words:最晚时间,location:{top:1190,left:343,width:141,height:37}},{words:周三深夜1125,location:{top:1251,left:345,width:307,height:51}},{words:回叠回,location:{top:1620,left:718,width:172,height:54}},{words:企业微信,location:{top:1756,left:184,width:146,height:44}},{words:分享到朋友圈,location:{top:2095,left:218,width:213,height:33}},{words:保存到相册,location:{top:2092,left:674,width:177,height:37}}],words_result_num:18,log_id:1797510538274524905}# pic_str json.loads(json_str)parse_pic_data(pic_orc_str)# main()部署流程
安装python_3.9
安装包有略过此过程
安装微信
先下载一个 3.9.10.19版本 32位的。登陆以后设置里面点击 更新升级到最新版然后拿23版本 覆盖一下 就ok了
安装python依赖
4. 安装依赖
sh
# 升级 pip
python -m pip install -U pip
# 安装必要依赖
pip install -r requirements_v1.txt
pip install baidu-aip
pip install pandas openpyxl
pip install xlrd
pip install pymem
运行微信 sh
python main.py# 需要停止按 CtrlC已支持功能
图片ocr识别添加人员
命令 添加人员 xxx xxx服务部 wxid_huwcf7p637mxxx
查看execl文件
命令数据文件 仅管理员权限的人支持
