python3 sqlite3 学习笔记（文件信息储存）

Cool_Breeze

[Python] 纯文本查看 复制代码

# coding=utf-8
# python3.7.9
# Cool_Breeze

import sqlite3
import os
import datetime
import time
import csv

# 获取文件信息
def getFileInfo(path):
    absname = ''
    timestamp = 0
    mdate = ''
    for rootDir, baseDir, files in os.walk(path):
        for file in files:
            absname = os.path.join(rootDir, file)
            timestamp = time.localtime(os.path.getmtime(absname))
            yield  (None,                                                    # id
                    file,                                                    # name
                    absname,                                                 # absname
                    datetime.datetime.strptime(
                    time.strftime('%Y%m%d%H%M%S',timestamp),'%Y%m%d%H%M%S'), # mdate
                    os.path.getsize(absname)                                 # size_KB
                    )

# 连接数据库
conn = sqlite3.connect('fileInfo.db')

# 游标
cur = conn.cursor()

# 删除表单
# cur.execute("drop table fileInfo")

# 创建表单
# id 自动增长的整数
# name str
# mdate datetime
# size_KB float

# cur.execute("""create table fileinfo(
    # id integer PRIMARY KEY,
    # name text,
    # absname text,
    # mdate date,
    # size_KB real)""")

# 插入数据
# for n in getFileInfo(r'D:\GIN\py'):
    # cur.execute("insert into fileInfo values(?,?,?,?,?)", n)

# 查询数据
# cur.execute("select name,size_KB from fileInfo where id=1000")
# print(cur.fetchone())

# 导出csv
# excel 打开乱码，需要将文件编码转换为 ANSI（源文件是 utf-8）
cur.execute("select * from fileInfo")
csvHeader = ['序号', '文件名', '文件绝对路径', '文件最后修改日期', '文件大小（单位：KB）']
with open('fileInfo.csv', 'w', encoding='utf-8',  newline='') as f:
    csvf = csv.writer(f)
    csvf.writerow(csvHeader)
    for n in cur.fetchall():
        csvf.writerow(n)

# 关闭游标
cur.close()
# 提交修改
conn.commit()
# 关闭连接
conn.close()

# 笔记

# 查询表名
# cur.execute("select name from sqlite_master WHERE  type='table'")
# tname = cur.fetchall()[0][0]
# print(tname)
# 查询表单字段名
# cur.execute(f"PRAGMA table_info({tname})")
# for n in cur.fetchall(): print(n)
# print(conn.total_changes)

# 创建表
# cur.execute('''create table test1(id integer, d date, ts timestamp)''')
# cur.execute("create table test1(id integer PRIMARY KEY, d date, ts timestamp)")
# 插入数据
# cur.execute("insert into test1 values(?,?,?)",(None,datetime.datetime.now(),time.time()))
# cur.execute("insert into test1 values(?,?,?)",(None,datetime.datetime.strptime('20180627041104','%Y%m%d%H%M%S'),time.time()))
# cur.execute("insert into test1 values(?,?,?)", (None,datetime.datetime.now(),time.time()))
# batch(cur)

# 查询表单数据
# cur.execute("select * from test1 where ts > 1613712545 or id %3= 0 ")
# for n in cur.fetchall():
    # print(n)

# sqlite3 内置函数
# cur.execute("select *, max(id) from test1")
# cur.execute("select sum(id) from test1")
# sqlite3 版本
# cur.execute("select sqlite_version()")
# target = cur.fetchone()
# cur.execute("select * from test1 where id=(?)", target)
# 日期函数
# cur.execute("select strftime('%Y-%m-%d %H:%M', 'now', '+8 hours')")

# 转换为时间戳
# cur.execute("select d from test1 where id=1")
# d = cur.fetchone()[0]
# cur.execute("select strftime('%s', '2021-02-21 08:02') - strftime('%s', ?)", (d,))
# cur.execute("select strftime('%', ?)", cur.fetchone())
# print(cur.fetchone())
# print(cur.fetchone())
# cur.execute("select ts from test1 where id=2")

# glob
# cur.execute("select id, name, mdate from fileinfo where name glob '*?.py'")

# like
# cur.execute("select id, name, mdate from fileinfo where name like '%_.py'")

# limit 输出限制
# cur.execute("select id, name, mdate from fileinfo where name like '%_.py' limit 2 offset 1")

# order by 排序  升序或降序 [ASC | DESC]
# 注意语句顺序 where --> group by --> order by --> limit
# cur.execute("select id, size_KB from fileinfo where name like '%_.py' order by size_KB asc limit 10")

# group by 数据分组
# cur.execute("select id,name,sum(size_KB)as size from fileinfo where name like '_.jpeg' group by name order by size asc")

# having 在 GROUP BY 子句创建的分组上设置筛选条件
# 注意语句顺序 where --> group by --> having --> order by --> limit
# 找出 以一个字符开头，以.jpeg结尾的文件名，函数count统计相同名出现的次数， 然后按相同名字分组，过滤掉出现次数小于300的， 按出现次数升序
# cur.execute("select id,name,count(name) as cn from fileinfo where name like '_.jpeg' group by name having cn>300 order by cn asc")

# distinct 去重
# cur.execute("select count(name) from fileinfo where name glob '?.jpeg'") # 找到 3034 个文件
# cur.execute("select count(distinct name) from fileinfo where name glob '?.jpeg'") # 去重后剩下 9 个文件


# for n in cur.fetchall():
    # print(n)
# 
# 更新数据
# cur.execute("select d from test1 where id=13")
# print(cur.fetchone())
# cur.execute("update test1 set d=(?) where id=13", (datetime.datetime.now(),))
# 删除表单
# cur.execute("drop table test1")

# delete语句
# cur.execute("delete from test1 where id=13")
# conn.commit()# saveCsv(cur)

Cool_Breeze

utf-8 转 gb18030 编码转换

[Python] 纯文本查看 复制代码

# coding=utf-8
# python3.7.9

# utf-8 转 GB18030
# str.encode(encoding="utf-8", errors="strict")

with open('fileInfo.csv', encoding='utf-8') as f:
    with open('fileInfoANSI.csv', 'wb') as af:
        for line in f:
            af.write(line.encode('GB18030', errors='ignore'))

Cool_Breeze

[Python] 纯文本查看 复制代码

import sqlite3
import hashlib
import os

def check_data_file():
    '''连接数据库'''
    db_name = "user.db"
    sql = None
    if not os.path.exists(db_name):
        sql = sqlite3.connect(db_name)
        cur = sql.cursor()
        cur.execute('''create table users(
            user text not null,
            password text not null)''')
    else:
        sql = sqlite3.connect(db_name)
    return sql


def hashlib_bytes(b_values):
    '''加密密码'''
    h = hashlib.blake2b()
    h.update(b_values)
    return h.hexdigest()

def register_user(sql_cur):
    '''注册用户'''
    while True:
        print("===========注册==============")
        input_user = input("请输入用户名:")
        print(input_user)
        if input_user == "0":break
        if input_user == "": continue
        sql_cur.execute(f"select user from users where user = '{input_user}'")
        if sql_cur.fetchall():
            print("用户已经存在")
            continue
        while True:
            input_user_passwd = input("请输入密码")
            if input_user_passwd == "":continue
            sql_cur.execute("insert into users values(?,?)", (input_user, hashlib_bytes(input_user_passwd.encode())))
            sql.commit()
            break
        break

def logon(sql_cur):
    '''登录'''
    while True:
        print("===========登录==============")
        input_user = input("请输入用户名:")
        if input_user == "0":break
        if input_user == "": continue
        sql_cur.execute(f"select user from users where user = '{input_user}'")
        if not sql_cur.fetchall():
            print(f"没有用户{input_user}")
            continue
        while True:
            input_user_passwd = input("请输入密码")
            if input_user_passwd == "":continue
            sql_cur.execute(f"select password from (select * from users where user = '{input_user}') where password = '{hashlib_bytes(input_user_passwd.encode())}'")
            if not sql_cur.fetchall():
                print(f"{input_user}登录失败！")
                continue
            print(f"{input_user}登录成功！")
            break
        break

def query(sql_cur):
    '''查询'''
    print("============查询==============")
    sql_cur.execute("select * from users")
    for n in sql_cur.fetchall():
        print(f"user：{n[0]}\tpassword：{n[1]}")

def update_password(sql_cur):
    '''修改密码'''
    while True:
        print("===========修改密码==============")
        input_user = input("请输入用户名:")
        print(input_user)
        if input_user == "0":break
        if input_user == "": continue
        sql_cur.execute(f"select user from users where user = '{input_user}'")
        if not sql_cur.fetchall():
            print(f"{input_user} 用户不存在")
            continue
        while True:
            input_user_passwd = input("请输入新密码")
            if input_user_passwd == "":continue
            sql_cur.execute(f"update users set password = '{hashlib_bytes(input_user_passwd.encode())}' where user = '{input_user}'")
            print("修改成功！")
            sql.commit()
            break
        break

def delete_user(sql_cur):
    '''删除用户'''
    while True:
        print("===========删除用户==============")
        input_user = input("请输入用户名:")
        print(input_user)
        if input_user == "0":break
        if input_user == "": continue
        sql_cur.execute(f"select user from users where user = '{input_user}'")
        if not sql_cur.fetchall():
            print(f"{input_user} 用户不存在")
            continue
        sql_cur.execute(f"delete from users where user = '{input_user}'")
        print(f"{input_user} 删除成功！")
        sql.commit()
        break

if __name__ == '__main__':
    sql = check_data_file()
    sql_cur = sql.cursor()
    while True:
        input_str = input("===========操作菜单==============\n0.退出\n1.登录\n2.注册\n3.查询用户\n4.修改密码\n5.删除用户\n请输入:")
        os.system("cls")
        if input_str == "0": break
        elif input_str == "1":
            logon(sql_cur)
        elif input_str == "2":
            register_user(sql_cur)
        elif input_str == "3":
            query(sql_cur)
        elif input_str == "4":
            update_password(sql_cur)
        elif input_str == "5":
            delete_user(sql_cur)
    sql.close()

netspirit

不会sqlite 每次我都用txt存储然后用正则表达式提取

Cool_Breeze

netspirit 发表于 2021-2-19 15:19
不会sqlite 每次我都用txt存储然后用正则表达式提取

数据库的性能应会高很多！

wzzjnb2006

这个要学习一下。