[PYTHON] 使用python实现一个miniWorkNote来记录工作内容

导读&需求

刚才做了啥事情? 忘了…

今天有变更吗? 不记得了…

对于我这种记忆力不好的人来说, 是需要一个工具来记录 已完成 和 待完成 的事情的. 比如一个记事本就行, 但我没法一眼就能找到还有哪些事情未做(都一个色儿). 那么就需要一个工具来帮我了.

要求:

1. 离线: 出于安全考虑
2. 使用简单: 不能比记事本复杂
3. 方便: 能一眼看出来还有哪些事情要做
4. 最好是有源码: 能简单的修改以实现更多的需求.
5. 免费: 穷! 这种软件花钱感觉不值得…
6. 安装简单: 不能有一大堆依赖包. 不能点下一步点到怀疑人生

好像/貌似没找到类似的软件. 这要求也太TM刁钻了. 谁提的需求? 原来是我自己, 那没事了.

这种情况, 作为一个MYSQL-DBA应该怎么办呢? 那当然是自己搓一个了卅!

设计&实现

语言: 直接选python, 成本最低. 改起来也简单, win/linux都兼容.
前端: 可以使用TK写界面, 但太丑了. 而且不熟. 那就使用 html吧
存储: 可以直接将py对象写入文件, 但兼容性不好, 也可以自己定义数据结构, 但对于修改操作太麻烦了. 所以直接使用sqlite3 (py也自带该接口)
交互: jq/ajax之类的太麻烦了, 直接XMLHttpRequest. 同理界面也使用bootstrap之类的, 稍微排版下即可(没得审美要求). 后端也不是有flask之类的, 直接走http server

为了方便使用, 这些功能就放一个文件就行.

作为一个完整的程序, 还得做到正常退出, 即处理SIGTERM之类的信号量(退出前要保存数据)

考虑到兼容性和数据恢复之类的, 我们也会记录DML语句到日志文件, 就相当于REDO LOG

本来打算做用户验证的, 但没必要, 就自己使用就行, 还安全点. 而且sqlite3也不支持多线程

我们的需求主要是记录已做/将做的事情 和 某个时间(确定)的变更. 所以可以设计两张表 work 和 plan.
work 记录已完成的工作和未完成的工作. 这个通常是没得时间限制的. 但涉及到状态变化, 即某件事情可能已经完成了, 也可能我先记录下, 但还没有完成. 工作内容可能也是会变化的. 所以要有编辑/修改功能
plan 是计划安排, 比如周五晚上做切换演练. 没有那么多状态, 完成后, 只需要反馈下即可.
DDL参考如下

create table work(
work_id bigint primary key, -- 使用时间戳作为ID吧(为了兼容性,就不使用comment了)
work_author varchar(255) default 'ddcw', -- 相关人员
work_dt datetime, -- 修改时间
work_status varchar(20), -- 完成/未完成/异常/已删除
work_si varchar(255), -- 简述
work_detail text, -- 详情
work_other varchar(255) -- 保留字段
);
create table plan(
plan_id bigint primary key,
plan_author varchar(255) default 'ddcw',
plan_dt datetime, -- 变更时间
plan_update datetime, -- 这行数据的修改时间, 可以当作是完成时间
plan_status varchar(20), -- 状态, 完成之后要手动设置为已完成. (已完成/未完成/异常)
plan_contact, -- 对接人
plan_si varchar(255), -- 任务内容
plan_detail varchar(255), -- 任务内容
plan_other varchar(255) -- 保留字段
);

最开始是设计登录选项的, 所以有author.

演示

原神启动:
我这里就用win环境演示吧. 除了要求python3, 没得其它任何要求的. 嘎嘎好用.
为了方便使用, html也是直接放在py里面的…

python miniWorkNote.py


然后浏览器打开那个地址, 就能看到如下信息(我这里有部分测试数据)

点击本/上周周报, 会自动复制相关内容到粘贴板

点击修改可以修改相关信息

添加也是可以的

其它的我就不测试了. 感兴趣的自己试一试.

总结

不使用框架/第三方包之类的, 写起来还是蛮舒服的(如果偶尔写写的话).
工具刚写完不多久, 目前用起来还不错(边写边用,边用边写), 后续好用的话再改进改进(感觉刷新按钮没啥用, 一直都是自动刷新的…).

一些中间遇到的坑, 前后端,数据库的都有…
前端: Date的月份得是从0开始的异世界. tr在元素附设置的背景颜色会被td(class)覆盖掉
后端: sqlite3不支持多线程, 也不支持comment

生成的.sql文件也是可以在mysql等数据库回放的. SQL版的REDO LOG

附源码

源码地址: https://github.com/ddcw/ddcw/tree/master/python/miniworknote

#!/usr/bin/env python
# write by ddcw https://github.com/ddcw
# 一个脚本, 用来记录做过的事情, 和安排.
# 只使用内置Python模块
# 理由: 外包,没得专门的系统来管理, 全靠记忆力....
# 只支持本地访问, 不支持远程访问(简单一点)
# 使用方法: python3 miniWorkNote.py 然后浏览器访问即可
# 数据存储在 .db (sqlite3) 文件, 最开始打算不使用数据库的, 但修改不方便, 就还是整了一个
# 还会保留一份SQL文件. 方便写入其它数据库
# 界面设计
"""
----NOTE------------------------PLAN----------
| worktime | 变更时间(计划时间) |
| 工作内存简述 | 变更内容 |
| 详情 | 对接人 |
| | |
| ADD FLUSH | ADD FLUSH |
-----------------------------------------------
| 本周周报 上周周报 |
-----------------------------------------------
| WORK 1 改|详|删 | PLAN 1 详|完成 |
| WORK 2 改|详|删 | PLAN 2 详|完成 |
| WORK 3 改|详|删 | PLAN 3 详|完成 |
| WORK 4 改|详|删 | |
"""
# CHAGE LOG
# at 2024.06.21 init
# at 2024.06.26 finish

# 变量名.
BIND_HOST = '127.0.0.1' # 监听的ip
BIND_PORT = 80 # 监听的端口
DB_FILE = 'miniWorkNote.db' # 保存数据的db文件, sqlite3
SQL_FILE = 'miniWorkNote.sql' # 保存数据的sql, 含DDL. like REDO
SAVE_DAYS = 3650000 # 保存天数. (还没实现...)

# 表结构定义
MINIWN_DDL = ["""
create table work(
work_id bigint primary key, -- 使用时间戳作为ID吧(为了兼容性,就不使用comment了)
work_author varchar(255) default 'ddcw', -- 相关人员
work_dt datetime, -- 修改时间
work_status varchar(20), -- 完成/未完成/异常/已删除
work_si varchar(255), -- 简述
work_detail text, -- 详情
work_other varchar(255) -- 保留字段
);
""","""
create table plan(
plan_id bigint primary key,
plan_author varchar(255) default 'ddcw',
plan_dt datetime, -- 变更时间
plan_update datetime, -- 这行数据的修改时间, 可以当作是完成时间
plan_status varchar(20), -- 状态, 完成之后要手动设置为已完成. (已完成/未完成/异常)
plan_contact, -- 对接人
plan_si varchar(255), -- 任务内容
plan_detail varchar(255), -- 任务内容
plan_other varchar(255) -- 保留字段
);
"""
]

import os,sys,signal
import datetime
import time
import sqlite3
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
import threading
import urllib.parse
import json

# 做全局初始化
if os.path.exists(SQL_FILE):
SQL_FD = open(SQL_FILE,'a')
else:
SQL_FD = open(SQL_FILE,'w')
# 更新下基础DDL
for ddl in MINIWN_DDL:
SQL_FD.write(ddl)
SQL_FD.write('\n') # 再来个换行, 好看点
SQL_FD.flush()

if os.path.exists(DB_FILE):
CONN = sqlite3.connect(DB_FILE)
else: # 不存在的话, 就初始化表结构
CONN = sqlite3.connect(DB_FILE)
for ddl in MINIWN_DDL:
cursor = CONN.cursor()
cursor.execute(ddl)
CONN.commit()

# 捕获15信号量, 做conn和fd的关闭
def signal_15_handler(sig,frame):
print("将自动退出")
CONN.commit()
CONN.close()
SQL_FD.flush()
SQL_FD.close()
print("已保存退出")
sys.exit(1)

signal.signal(signal.SIGTERM, signal_15_handler) # kill -15
signal.signal(signal.SIGINT, signal_15_handler) # ctrl+c

def runsql(sql):
try:
cursor = CONN.cursor()
cursor.execute(sql)
data = cursor.fetchall()
status = True
sql += f"{';' if data[-1:] != ';' else ''}" + "\n"
if sql[:7].upper() != "SELECT ":
SQL_FD.write(sql)
except Exception as e:
status = False
data = str(e)
sql = "-- /* MAY BE ERROR " + data + " */" + sql + f"{';' if data[-1:] != ';' else ''}" + "\n"
SQL_FD.write(sql)
SQL_FD.flush()
return status,data

# 啊哈! sqlite3不支持多线程
#class ThreadingSimpleServer(ThreadingMixIn, HTTPServer):
# pass

class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
url_components = urllib.parse.urlparse(self.path)
query = urllib.parse.parse_qs(url_components.query)
path = url_components.path
if path == '/work':
#self.handle_work_request(query)
self.handle_html_request()
elif path == '/plan':
#self.handle_work_plan(query)
self.handle_html_request()
else:
self.handle_html_request()

def do_POST(self):
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
data = json.loads(post_data)
print('PATH',self.path,"收到的数据:",data)
if self.path == '/work':
# 判断action
if data['action'] == 'add': # 新增
if data['work_dt'] == '':
data['work_dt'] = str(datetime.datetime.now()).split('.')[0]
sql = f"insert into work values({int(time.time()*1000)},'ddcw',\'{data['work_dt']}\',\'{data['work_status']}\', \'{data['work_si']}\', \'{data['work_detail']}\', '')"
elif data['action'] == 'delete': # 删除
# sql = f"delete from work where work_id={data['work_id']}"
sql = f"update work set work_status='已删除' where work_id={data['work_id']}"
elif data['action'] == 'update': # 更新
sql = f"update work set work_status=\'{data['work_status']}\', work_si=\'{data['work_si']}\', work_dt='{str(datetime.datetime.now()).split('.')[0]}' where work_id = {data['work_id']}"
elif data['action'] == 'select': # 查询的, 默认就最近50条
sql = 'select work_id,work_dt,work_status,work_si,work_detail from work where work_status !="已删除" order by work_dt desc limit 50'
else:
sql = "select '不知道你在干什么, 但我只管记录... {data}'"
#self.wfile.write(post_data) # 慕容复
elif self.path == '/plan':
# 判断action
if data['action'] == 'add':
if data['plan_dt'] == '':
data['plan_dt'] = str(datetime.datetime.now()).split('.')[0]
sql = f"insert into plan values({int(time.time()*1000)}, 'ddcw', \'{data['plan_dt']}\', \'{str(datetime.datetime.now()).split('.')[0]}\', '未完成', \'{data['plan_contact']}\', \'{data['plan_si']}\', \'{data['plan_detail']}\', '' )"
elif data['action'] == 'status':
sql = f"update plan set plan_status=\'{data['plan_status']}\',plan_update=\'{str(datetime.datetime.now()).split('.')[0]}\' where plan_id={data['plan_id']}"
elif data['action'] == 'select':
sql = f"select plan_id,plan_dt,plan_contact,plan_status,plan_si,plan_detail from plan where plan_status='未完成' order by plan_dt" # 就不要limit了, 全TM查出来
else:
sql = "select '不知道你在干什么, 但我只管记录... {data}'"
elif self.path == '/report':
if data['action'] == 1: # 上周周报
sql = f"select work_id,work_dt,work_si from work where work_dt >= date('now', 'weekday 0', '-13 days') and work_dt = date('now', 'weekday 0', '-6 days') and work_dt < date('now', 'weekday 0', '+1 day') and work_status!='已删除'"
else:
sql = "select '不知道你在干什么, 但我只管记录... {data}'"
status,data = runsql(sql)
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
rbdata = json.dumps({'data':data,'status':status}).encode('utf-8')
print('返回数据',{'data':data,'status':status})
self.wfile.write(rbdata)

def handle_html_request(self):
#pass # 就返回首页html就行. 其实就这一个page...
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
html_content = '''

DDCW miniWorkNote

NOTE

任务时间:
任务状态 未完成已完成异常

工作内容简述:
工作内容详情:

任务ID

修改时间

任务状态

任务简述

操作

PLAN

任务时间:
对接人:

工作内容简述:
工作内容详情:

ID

时间

对接人

状态

简述

操作

'''
self.wfile.write(html_content.encode('utf-8'))

def handle_work_request(self,query):
pass

def handle_work_plan(self,query):
pass

#def run(server_class=ThreadingSimpleServer, handler_class=SimpleHTTPRequestHandler):
def run(server_class=HTTPServer, handler_class=SimpleHTTPRequestHandler):
server_address = (BIND_HOST,BIND_PORT)
httpd = server_class(server_address, handler_class)
print(f'http://{BIND_HOST}:{BIND_PORT}')
httpd.serve_forever()

if __name__ == '__main__':
run() # 润