99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
import sys
|
||
from datetime import date, timedelta
|
||
|
||
import cv2
|
||
from sqlalchemy import update, and_, select, exists
|
||
|
||
from db import MysqlSession
|
||
from db.mysql import ZxIeOcrerror, ZxPhrec
|
||
from photo_mask.auto_photo_mask import mask_photo
|
||
from ucloud import ufile
|
||
from util import image_util, util
|
||
|
||
|
||
def check_error(error_ocr):
|
||
img_url = ufile.get_private_url(error_ocr.cfjaddress, "drg2015")
|
||
if not img_url:
|
||
# 没有自动涂抹的图片
|
||
img_url = ufile.get_private_url(error_ocr.cfjaddress, "drg103")
|
||
if not img_url:
|
||
raise Exception("云空间已删除")
|
||
name = error_ocr.cXm
|
||
id_card_num = error_ocr.cSfzh
|
||
|
||
image = mask_photo(img_url, name, id_card_num, (0, 0, 0))[1]
|
||
final_img_url = ufile.get_private_url(error_ocr.cfjaddress, "drg100")
|
||
final_image, _ = image_util.read(final_img_url)
|
||
return image_util.combined(final_image, image)
|
||
|
||
|
||
def auto_check_error(error_ocr):
|
||
if error_ocr.cfjaddress2:
|
||
return error_ocr.cfjaddress2
|
||
|
||
cfjaddress = error_ocr.cfjaddress
|
||
if cfjaddress[17] == "1":
|
||
return "图片类型错误"
|
||
|
||
session = MysqlSession()
|
||
cfjaddress = cfjaddress.rsplit(".", 1)[0] + ".%"
|
||
query = select(exists().where(ZxPhrec.cfjaddress.like(cfjaddress)))
|
||
record_exists = session.execute(query).scalar()
|
||
session.close()
|
||
if not record_exists:
|
||
return "未同步"
|
||
|
||
return "未知错误"
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 默认
|
||
today = date.today()
|
||
session = MysqlSession()
|
||
ocr_error = (session.query(ZxIeOcrerror.pk_phrec, ZxIeOcrerror.cXm, ZxIeOcrerror.cSfzh, ZxIeOcrerror.cfjaddress,
|
||
ZxIeOcrerror.cfjaddress2)
|
||
.filter(and_(ZxIeOcrerror.checktime.is_(None),
|
||
# ZxIeOcrerror.creationtime >= today,
|
||
ZxIeOcrerror.creationtime < today + timedelta(days=1)))
|
||
.limit(1).one())
|
||
session.close()
|
||
# 手动填充
|
||
# ocr_error = ZxIeOcrerror()
|
||
# ocr_error.pk_phrec = 0
|
||
# ocr_error.cXm = ""
|
||
# ocr_error.cSfzh = ""
|
||
# ocr_error.cfjaddress = ""
|
||
# ocr_error.cfjaddress2 = ""
|
||
# 手动选择pk
|
||
# session = MysqlSession()
|
||
# ocr_error = (session.query(ZxIeOcrerror.pk_phrec, ZxIeOcrerror.cXm, ZxIeOcrerror.cSfzh, ZxIeOcrerror.cfjaddress,
|
||
# ZxIeOcrerror.cfjaddress2)
|
||
# .filter(ZxIeOcrerror.pk_phrec == 0).one())
|
||
# session.close()
|
||
|
||
error_descript = auto_check_error(ocr_error)
|
||
try:
|
||
img = check_error(ocr_error)
|
||
cv2.imwrite(f"./mask_error_check/{ocr_error.cfjaddress}.jpg", img)
|
||
except Exception as e:
|
||
error_descript = str(e)
|
||
|
||
if error_descript == "未知错误":
|
||
while True:
|
||
check_finish = input("是否完成(是/否)：")
|
||
if check_finish == "是":
|
||
error_descript = input("错误描述：")
|
||
break
|
||
elif check_finish == "否":
|
||
sys.exit()
|
||
else:
|
||
print("无效输入！请输入“是”或者“否”")
|
||
|
||
session = MysqlSession()
|
||
update_error = (update(ZxIeOcrerror).where(ZxIeOcrerror.pk_phrec == ocr_error.pk_phrec).values(
|
||
checktime=util.get_default_datetime(), cfjaddress2=error_descript))
|
||
session.execute(update_error)
|
||
session.commit()
|
||
session.close()
|
||
print(f"【{ocr_error.pk_phrec}】错误分析已完成。错误描述：{error_descript}")
|