1. 安装依赖项首先，在开始编写代码之前，我们需要确保安装了必要的依赖项。我们需要安装的主要依赖项是ebooklib和beautifulsoup4。可以使用如下命令安装它们：pip install ebooklib beautifulsoup42. 示例脚本下面是一个简单的Python脚本，实现了将EPUB文件转换为TXT文件的功能。我们将函数划分为两部分，epub_to_txt() 负责将EPUB文件内容提取为文本，save_to_txt() 则负责将提取到的文本内容保存到一个TXT文件。import os import sys from ebooklib import epub from bs4 import BeautifulSoup def epub_to_txt(epub_path): book = epub.read_epub(epub_path) output_txt = "" for item in book.get_items(): if item.media_type == 'application/xhtml+xml': content = item.get_content() soup = BeautifulSoup(content, features="html.parser") text = soup.get_text() output_txt += text + "\n" return output_txt def save_to_txt(file_path, content): with open(file_path, 'w', encoding='utf-8') as txt_file: txt_file.write(content) if __name__ == '__main__': # epub文件路径和输出txt文件路径 epub_file = sys.argv[1] txt_file = sys.argv[2] # 从epub文件中提取文本 txt_content = epub_to_txt(epub_file) # 将文本保存到txt文件 save_to_txt(txt_file, txt_content) 3. 如何使用要使用上述脚本，您只需将其保存为.py文件，例如epub_to_txt.py。然后在命令行界面运行以下命令：python epub_to_txt.py <input_epub_file> <output_txt_file>将<input_epub_file>替换为要转换的EPUB文件的路径，<output_txt_file>替换为要生成的TXT文件的路径。之后，脚本将会将EPUB文件转换为TXT文件，并将结果保存在指定的输出文件中。4. 总结这篇博客提供了一个简单、轻量级的解决方案，使用Python和第三方库EbookLib快速将EPUB文件转换为TXT文件。
记录一次迁移MySQL大体积数据表的实践
近期学习实践数据库的笔记，文章内容和代码均系原创（部分知识概述型文本由soarchat 1.4和gpt4提供成文支持），转载请注明来源！前言近期因某线上场景特殊需要，得对一张6.19GB的MySQL数据表进行数据库环境的迁移，首先考虑到的方法当然是通过手头Navicat自带的导入/导出/转储功能来快速实现。但是经过半天的尝试，不管通过什么样的文件结构作为媒介，在导入的过程中总会发生一些错误问题（最好的一次还丢失了300多条数据）。在按照这种思路折腾的过程中，还发生了几次电脑的内存溢出而不得不重启OwO...于是尝试在网上搜寻关于这种问题的解决方案，谁料网上也是一片哀鸿遍野。个别能解决的，要么是让把文件导入到Linux系统中，通过Linux独有的split命令先分割再进行操作完后再取出，亦或是修改MySQL的类似于max_allowed_packet的各种配置（亲测无效），再或者就是换了其他的数据库管理工具（mysqldump之类的）。见到此状，想到了一句名言“自己动手，丰衣足食”，说干就干，在充分考虑读取大文件对内存的需求的基础上，对程序进行了严格的内存优化。最终的结果是一次执行通过，零错误零丢失，同时也没有发生内存泄漏！兴奋之余决定把总体的设计思路和程序开源到这篇文章里面，供后续有需求的人进行参考或修改/优化。如有大佬有更好的思路或优化建议，还望不吝赐教！！ ::twemoji:prayer::思路导出数据正常走Navicat的转储流程没啥问题，得到一个体积超大的sql文件之后，如何把它执行到一个新的数据库里是接下来讨论的重点问题。STEP1：拆分splitsql.py本程序的主要目标是高效地将大型SQL文件拆分成更小的文件。核心原理在于使用带缓冲区的文件读写操作，通过计数器和指定的行数限制将源文件逐行划分为多个小文件。关键步骤：1.逐行读取输入文件2.将每一行内容添加到一个临时缓存列表中，直到达到预先设定的行数限制3.新建一个带有编号的输出文件，将缓存的数据写入该文件。重复这个过程，直到遍历完整个输入文件。4.程序最后会判断是否还有剩余未处理的数据，若有，则将剩余数据写入一个新的输出文件。意义：提供了一种简洁高效的方法来处理大型SQL文件，尤其在数据库结构复杂、资源有限的场景下格外实用。此外，通过自定义行数限制和文件名前缀，用户可以根据实际需求对程序进行定制。总之，这是一个有助于简化大型数据库的维护和管理的工具。input_file_path = "table_testdata.sql" num_lines_per_file = 500 output_file_prefix = "" def split_sql_file(): with open(input_file_path, "r", encoding="utf-8", errors="ignore", buffering=8192) as input_file: file_count = 1 line_counter = 0 data = [] for line in input_file: if line_counter < num_lines_per_file: data.append(line) line_counter += 1 else: with open(f"{output_file_prefix}{file_count}.sql", "w", encoding="utf-8", buffering=8192) as output_file: output_file.writelines(data) file_count += 1 line_counter = 1 data = [line] if data: with open(f"{output_file_prefix}{file_count}.sql", "w", encoding="utf-8", buffering=8192) as output_file: output_file.writelines(data) if __name__ == '__main__': split_sql_file() STEP2：重组importsql.py程序主要功能是自动执行指定目录下的所有SQL文件。程序的核心原理在于通过连接到MySQL数据库，然后顺序执行目录中按文件名排序的SQL文件。程序首先通过配置信息创建一个数据库连接，并处理可能出现的错误。之后，它会遍历目录下所有以“.sql”为扩展名的文件，将其内容逐个加载并划分为SQL语句。为了优化批量执行效率，它将SQL语句分组成批次，并在执行过程中定时检查数据库连接是否仍然可用。该程序实现了以下功能：通过设置可配置参数及错误处理，确保连接及执行过程更为可靠；在程序每次执行之间设置延时，从而减轻数据库压力；以另外提供的log.txt文件形式记录错误和执行情况，方便追踪。这款程序可以节省数据库管理过程中的时间与手动执行脚本的工作量，提高工作效率，尤其适用于需要定期执行大量SQL脚本的场景。import os import mysql.connector import logging import time from mysql.connector import errorcode STATEMENTS_PER_BATCH = 1000 TIME_BETWEEN_BATCHES = 0 logging.basicConfig(filename='log.txt', level=logging.INFO, filemode='w', format='%(asctime)s - %(message)s') def connect_to_database(): config = { 'user': 'test', 'password': 'test', 'host': 'localhost', 'database': 'test', 'connection_timeout': 86400, } try: cnx = mysql.connector.connect(**config) return cnx except mysql.connector.Error as err: if err.errno == errorcode.ER_ACCESS_DENIED_ERROR: logging.error("有错误发生：用户名或密码错误") elif err.errno == errorcode.ER_BAD_DB_ERROR: logging.error("有错误发生：数据库不存在") else: logging.error(err) return None def execute_sql_files_in_directory(cnx): cursor = cnx.cursor() sql_files = [f for f in os.listdir('.') if f.endswith('.sql')] # 定义一个函数，用于从文件名中获取数字，并返回一个整数 def get_file_number(file_name): import re match = re.search(r'\d+', file_name) # 查找文件名中的数字 if match: return int(match.group()) # 将数字转换为整数并返回 else: return 0 # 如果没有找到数字，返回 0 # 使用 sorted 函数对 sql_files 列表进行排序，并传入 get_file_number 函数作为关键字参数 sql_files = sorted(sql_files, key=get_file_number) for sql_file in sql_files: logging.info(f"正在执行 {sql_file} ...") with open(sql_file, 'r', encoding='utf-8') as file: sql_script = file.read() sql_statements = sql_script.split(';') statement_batches = [sql_statements[i:i + STATEMENTS_PER_BATCH] for i in range(0, len(sql_statements), STATEMENTS_PER_BATCH)] # 在执行每个 SQL 文件之前检查连接是否存活，如果不存活则重新连接 try: cnx.ping(reconnect=True) logging.info(f"连接正常") except mysql.connector.Error as err: logging.error(f"连接失败：{err}") break for batch_idx, batch in enumerate(statement_batches): for idx, statement in enumerate(batch, start=batch_idx * STATEMENTS_PER_BATCH + 1): if statement.strip(): try: cursor.execute(statement) cnx.commit() logging.info(f"{sql_file}（语句 {idx}）执行成功！") except mysql.connector.Error as err: logging.error(f"执行 {sql_file}（语句 {idx}）时出错：{err}") break except Exception as e: logging.error(f"执行 {sql_file}（语句 {idx}）时出现未知错误：{str(e)}") break time.sleep(TIME_BETWEEN_BATCHES) cursor.close() def main(): cnx = None # 添加一个 while 循环，用于在连接失败时重试连接，直到成功或达到最大重试次数为止 max_retries = 5 retries = 0 while cnx is None and retries < max_retries: cnx = connect_to_database() if cnx is None: logging.info(f"连接失败，正在重试（{retries + 1}/{max_retries}）...") retries += 1 time.sleep(5) else: logging.info(f"连接成功") break if cnx is not None: execute_sql_files_in_directory(cnx) cnx.close() else: logging.error(f"无法连接到数据库") if __name__ == '__main__': main() 参考资料：MySQL mysqldump数据导出详解 - pursuer.chen - 博客园MySQL快速导入千万级别的大数据量sql文件 - 知乎MySql导入和抽取大数量级文件数据 - 知乎
问题描述：import pygame后每次运行py都会输出以下提示信息：pygame 2.4.0 (SDL 2.26.4, Python 3.8.6) Hello from the pygame community. https://www.pygame.org/contribute.html解决方案：参考资料：Python3.7如何解决Pygame显示欢迎的问题_注释pygame的欢迎_tiantian520_tt的博客-CSDN博客
本文由 soarAI 1.1 驱动，转载请注明来源。Ubuntu flask后台静默运行您可以使用以下命令在Ubuntu上使用Flask应用程序进行后台静默运行：nohup python3 your_flask_app.py > /dev/null 2>&1 &其中，your_flask_app.py是您的Flask应用程序的文件名。此命令将在后台启动Flask应用程序，并将输出重定向到/dev/null以避免在终端中显示输出。如果您想要停止Flask应用程序，请使用以下命令：ps aux | grep your_flask_app.py找到Flask应用程序的进程ID，然后使用以下命令杀死进程：kill -9 <process_id>其中，<process_id>是Flask应用程序的进程ID。
本文由 soarAI 1.1 驱动，转载请注明来源。ubuntu 更新python3您可以使用以下命令来更新Ubuntu上的Python 3：sudo apt-get update sudo apt-get install python3这将会更新您的系统上的Python 3版本到最新。如果您想要安装特定版本的Python 3，可以使用以下命令：sudo apt-get install python3.x其中，x代表您想要安装的Python 3版本号。例如，如果您想要安装Python 3.8，可以使用以下命令：sudo apt-get install python3.8sudo apt-get install python3.8执行后python3 -v仍然是3.6如果您已经安装了Python 3.8，但是在运行python3 -V时仍然显示版本为3.6，可能是因为系统中存在多个Python版本，并且3.6是默认版本。您可以尝试使用以下命令来更改默认Python版本：sudo update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.8 1这将会将Python 3.8设置为默认版本。然后，您可以再次运行python3 -V来验证版本是否已经更改。
前言今天无意间在网上发现了一款Windows系统可用、开源、面向普通用户、基于PaddleOCR的本地文本识别工具。经过对它的一系列配置，将本地的配置文件导出并记录在这里，方便日后在其他电脑上进行快速部署。PaddleOCR是一款基于PaddlePaddle的轻量级开源文本识别模型库，今年年初为满足毕设中证件智能识别模块开发需求做过一些研究（这里是当时对PaddleOCR的配置使用笔记）。没想到竟然能在今年年末发现有大佬从今年3月底开始同样基于这款模型库开发并开源了一款面向用户的，方便快捷的文本识别工具！相信很多人和我一样，经常会有在电脑上进行OCR操作的需求。从几年前开始我陆续使用过很多第三方的工具（如：天若OCR、QQ截图OCR等），但是它们终究都没能很好的满足我在体验方面的需求。于是自己开发了一个基于Web异步请求和百度高精接口的轻量级应用，使用范围也局限于自己和身边的个别人。毕竟使用别人的在线接口，还是会有很多不稳定性因素在里面的。原本打算年底在自己服务器上基于PaddleOCR server端模型对该应用进行重构，现在看来已经必要性不大了。Umi-OCR简介按照官方的介绍，这是一款免费，开源，可批量的离线OCR软件：兼容性系统支持 Win10 x64 及以上版本。CPU必须具有AVX指令集。出现初始化引擎失败等问题时请参考 问题排除 。地址Github地址Github Release v1.3.2地址官方提供的蓝奏云地址soarli备用地址更新日志（截至12.8）v1.3.2 2022.12.1 新功能：创建开机启动项时，可选不显示主窗口。 新功能：OCR结果输出到每个图片同名的单独txt文件。 新功能：增加独立的设置语言窗口，可在多处点开，便于切换语言。 新功能：合并段落添加合并自然段-西文模式，可在英文段落换行时补充空格。 新功能：快捷识图可选自动清空面板，只显示本次识别结果，且隐藏时间信息。 修复了一些BUG。 v1.3.1 2022.11.4 修Bug：快捷键模块重写，引入pynput库，舍弃keyboard库，解决几率失效、录制不正确等Bug。 新功能：添加开机自启，桌面快捷方式，开始菜单快捷方式。 新功能：多开软件时提示。 新功能：截图时隐藏窗口。 调整UI：使用频率极低的设置项设为隐藏的高级选项。 优化：检查引擎组件是否存在。 优化：横排-合并多行-自然段 优化逻辑，支持0~2全角空格首行缩进。 v1.3.0 2022.9.29 新功能：框选截屏。 新功能：系统托盘图标。 新功能：引擎进程常驻。 新功能：文本块后处理模块。 新功能：自定义主输出栏字体。 新功能：设置窗口弹出模式（保持置顶）。 调整UI：自适应Win风格组件。 修正了Bug：系统语言兼容性问题 issue #16 。 修正了Bug：微信图片粘贴问题 issue #22 。 更新PaddleOCR-json模块至v1.2.1，提供剪贴板支持。快捷识图通过剪贴板中转，无需再保存临时文件到硬盘。 v1.2.6 2022.9.1 更新PaddleOCR-json模块至v1.2.0，提高识别速度、准确度。 调整UI：更方便地用下拉框切换识别语言。 调整UI：可以从主窗口任意位置/任意选项卡拖入图片。 修正了Bug：提高程序健壮性，增加启动子进程时的更多异常处理情况。 修正了Bug：彻底解决了对边缘过窄的图片，识别结果不准确的问题 issue #7 。 优化适配PP-OCRv3模型，彻底解决了v3版模型比v2慢、不准的问题 issue #4 。 v1.2.5 2022.7.22 新功能：计划任务。识图完成后执行自动关机等任务。 新功能：可选拖入文件夹时递归导入子文件夹中所有图片。 调整UI：添加一些配置文件的快捷入口。 v1.2.4 2022.6.4 新功能：可选识别剪贴板图片后自动复制识别的文本。 补充功能：快捷键调用剪贴板识图时，若程序窗口被最小化，则恢复前台状态并挪到最前位置。 v1.2.3 2022.5.31 新功能：读取剪贴板图片。配置全局快捷键调用该功能。 v1.2.2 2022.4.30 新功能：可选任务完成后自动打开输出文件或目录。 v1.2.1 2022.4.16 更新PaddleOCR-json模块至v1.1.1，修正了可能得到错误包围盒的漏洞。 v1.2.0 2022.4.8 可选生成图文链接.md文件，作为索引使用有更佳的观感。 修改设置面板的样式，改为滚动面板以容纳更多设置选项。 用户修改配置项后可自动保存。 v1.1.1 2022.3.30 修正了Bug：退出忽略区域窗口时，OCR子进程未关闭。 v1.1.0 2022.3.30 新功能：忽略区域窗口以虚线框 展示识别出的文字块。 v1.0.0 2022.3.28 “梦开始的地方”体验综合体验还是很好的，文本识别准确率也比较高（就是界面丑了点，哈哈，问题不大）。用户界面（按照我的配置，几乎用不到）识别内容：设置：功能针对批量图片需要处理的场景功能也很强大，在这里我就不一一展开介绍了，可以参考这里。配置备份配置后通过win+shift+a快捷键来触发截图，截图识别结果将不再主界面展示，而是直接被复制到剪切板，主界面“识别内容”可作为识别历史记录加以回溯参考。{ "isDebug": false, "isAdvanced": false, "isTray": true, "isBackground": true, "clickTrayModeName": "显示面板", "textpanelFontFamily": "Microsoft YaHei UI", "textpanelFontSize": 14, "isTextpanelFontBold": false, "WindowTopMode": 0, "isAutoStartup": true, "isStartMenu": true, "isDesktop": false, "isHotkeyClipboard": true, "hotkeyClipboard": "win+alt+v", "isHotkeyScreenshot": true, "isScreenshotHideWindow": false, "screenshotHideWindowWaitTime": 200, "hotkeyScreenshot": "shift+win+a", "hotkeyMaxTtl": 2.0, "isHotkeyStrict": false, "scsModeName": "Umi-OCR 软件截图", "scsColorLine": "#3366ff", "scsColorBoxUp": "#000000", "scsColorBoxDown": "#ffffff", "isNeedCopy": true, "isNeedClear": false, "isOpenExplorer": false, "isOpenOutputFile": false, "okMissionName": "关机", "okMission": { "关机": { "code": "msg %username% /time:25 \"Umi-OCR任务完成，将在30s后关机\" & echo 关闭本窗口可取消关机 & choice /t 30 /d y /n >nul & shutdown /f /s /t 0" }, "休眠": { "code": "msg %username% /time:25 \"Umi-OCR任务完成，将在30s后休眠\" & echo 关闭本窗口可取消休眠 & choice /t 30 /d y /n >nul & shutdown /f /h" } }, "isRecursiveSearch": false, "isOutputTxt": true, "isOutputSeparateTxt": false, "isOutputMD": false, "isOutputJsonl": false, "isIgnoreNoText": true, "tbpuName": "优化单行", "isAreaWinAutoTbpu": false, "ocrToolPath": "PaddleOCR-json/PaddleOCR_json.exe", "ocrRunModeName": "", "ocrConfigName": "简体中文", "ocrConfig": { "简体中文": { "path": "PaddleOCR_json_config_ch.txt" } }, "argsStr": "", "isOcrAngle": false, "ocrCpuThreads": 10, "isOcrMkldnn": true, "ocrLimitModeName": "长边压缩模式", "ocrLimitSize": 960, "imageSuffix": ".jpg .jpe .jpeg .jfif .png .webp .bmp .tif .tiff", "isLanguageWinAutoExit": false, "isLanguageWinAutoOcr": false, "processID": 3128, "processKey": "1670492176.8567443", "promptScreenshotScale": true, "promptMultiOpen": true }参考资料：完全免费，不用联网，这套OCR工具太好用了hiroi-sora · GitHubRelease Umi-OCR 文字识别工具 v1.3.2 · hiroi-sora/Umi-OCR · GitHub
本文转载自互联网，已在文末注明来源。1、redis连接redis提供两个类Redis和StrictRedis用于实现Redis的命令，StrictRedis用于实现大部分官方的命令，并使用官方的语法和命令，Redis是StrictRedis的子类，用于向后兼容旧版本的redis-py。redis连接实例是线程安全的，可以直接将redis连接实例设置为一个全局变量，直接使用。如果需要另一个Redis实例（or Redis数据库）时，就需要重新创建redis连接实例来获取一个新的连接。同理，python的redis没有实现select命令。安装redispip install redis连接redis，加上decode_responses=True，写入的键值对中的value为str类型，不加这个参数写入的则为字节类型。import redis # 导入redis模块，通过python操作redis 也可以直接在redis主机的服务端操作缓存数据库 r = redis.Redis(host='localhost', port=6379, decode_responses=True) # host是redis主机，需要redis服务端和客户端都启动 redis默认端口是6379 r.set('name', 'junxi') # key是"foo" value是"bar" 将键值对存入redis缓存 print(r['name']) print(r.get('name')) # 取出键name对应的值 print(type(r.get('name')))2、连接池redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池连接池import redis # 导入redis模块，通过python操作redis 也可以直接在redis主机的服务端操作缓存数据库 pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) # host是redis主机，需要redis服务端和客户端都起着 redis默认端口是6379 r = redis.Redis(connection_pool=pool) r.set('gender', 'male') # key是"gender" value是"male" 将键值对存入redis缓存 print(r.get('gender')) # gender 取出键male对应的值3、redis基本命令 Stringset(name, value, ex=None, px=None, nx=False, xx=False)在Redis中设置值，默认，不存在则创建，存在则修改参数：ex，过期时间（秒）px，过期时间（毫秒）nx，如果设置为True，则只有name不存在时，当前set操作才执行xx，如果设置为True，则只有name存在时，当前set操作才执行1.ex，过期时间（秒） 这里过期时间是3秒，3秒后p，键food的值就变成Noneimport redis pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) r.set('food', 'mutton', ex=3) # key是"food" value是"mutton" 将键值对存入redis缓存 print(r.get('food')) # mutton 取出键food对应的值2.px，过期时间（豪秒） 这里过期时间是3豪秒，3毫秒后，键foo的值就变成Noneimport redis pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) r.set('food', 'beef', px=3) print(r.get('food'))3.nx，如果设置为True，则只有name不存在时，当前set操作才执行 （新建）import redis pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) print(r.set('fruit', 'watermelon', nx=True)) # True--不存在 # 如果键fruit不存在，那么输出是True；如果键fruit已经存在，输出是None4.xx，如果设置为True，则只有name存在时，当前set操作才执行 （修改）print((r.set('fruit', 'watermelon', xx=True))) # True--已经存在 # 如果键fruit已经存在，那么输出是True；如果键fruit不存在，输出是None5.setnx(name, value)设置值，只有name不存在时，执行设置操作（添加）print(r.setnx('fruit1', 'banana')) # fruit1不存在，输出为True6.setex(name, time, value) 设置值参数：time，过期时间（数字秒 或 timedelta对象）import redis import time pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) r.setex("fruit2", 5, "orange") time.sleep(5) print(r.get('fruit2')) # 5秒后，取值就从orange变成None7.psetex(name, time_ms, value)设置值参数：time_ms，过期时间（数字毫秒 或 timedelta对象）r.psetex("fruit3", 5000, "apple") time.sleep(5) print(r.get('fruit3')) # 5000毫秒后，取值就从apple变成None8.mset(args, *kwargs)批量设置值如：r.mget({'k1': 'v1', 'k2': 'v2'}) r.mset(k1="v1", k2="v2") # 这里k1 和k2 不能带引号 一次设置对个键值对 print(r.mget("k1", "k2")) # 一次取出多个键对应的值 print(r.mget("k1"))9.mget(keys, *args)批量获取如：print(r.mget('k1', 'k2')) print(r.mget(['k1', 'k2'])) print(r.mget("fruit", "fruit1", "fruit2", "k1", "k2")) # 将目前redis缓存中的键对应的值批量取出来10.getset(name, value)设置新值并获取原来的值print(r.getset("food", "barbecue")) # 设置的新值是barbecue 设置前的值是beef11.getrange(key, start, end)获取子序列（根据字节获取，非字符）参数：name，Redis 的 namestart，起始位置（字节）end，结束位置（字节）如： "君惜大大" ，0-3表示 "君"r.set("cn_name", "君惜大大") # 汉字 print(r.getrange("cn_name", 0, 2)) # 取索引号是0-2 前3位的字节 君 切片操作 （一个汉字3个字节 1个字母一个字节 每个字节8bit） print(r.getrange("cn_name", 0, -1)) # 取所有的字节 君惜大大 切片操作 r.set("en_name","junxi") # 字母 print(r.getrange("en_name", 0, 2)) # 取索引号是0-2 前3位的字节 jun 切片操作 （一个汉字3个字节 1个字母一个字节 每个字节8bit） print(r.getrange("en_name", 0, -1)) # 取所有的字节 junxi 切片操作12.setrange(name, offset, value)修改字符串内容，从指定字符串索引开始向后替换（新值太长时，则向后添加） 参数：offset，字符串的索引，字节（一个汉字三个字节）value，要设置的值r.setrange("en_name", 1, "ccc") print(r.get("en_name")) # jccci 原始值是junxi 从索引号是1开始替换成ccc 变成 jccci13.setbit(name, offset, value)对name对应值的二进制表示的位进行操作参数：name，redis的nameoffset，位的索引（将值变换成二进制后再进行索引）value，值只能是 1 或 0注：如果在Redis中有一个对应： n1 = "foo"， 那么字符串foo的二进制表示为：01100110 01101111 01101111 所以，如果执行 setbit('n1', 7, 1)，则就会将第7位设置为1， 那么最终二进制则变成 01100111 01101111 01101111，即："goo" 扩展，转换二进制表示： source = "陈思维" source = "foo" for i in source: num = ord(i) print bin(num).replace('b','') 特别的，如果source是汉字 "陈思维"怎么办？ 答：对于utf-8，每一个汉字占 3 个字节，那么 "陈思维" 则有 9个字节 对于汉字，for循环时候会按照 字节 迭代，那么在迭代时，将每一个字节转换 十进制数，然后再将十进制数转换成二进制 11100110 10101101 10100110 11100110 10110010 10011011 11101001 10111101 1001000014.getbit(name, offset)获取name对应的值的二进制表示中的某位的值 （0或1）print(r.getbit("foo1", 0)) # 0 foo1 对应的二进制 4个字节 32位 第0位是0还是115.bitcount(key, start=None, end=None)获取name对应的值的二进制表示中 1 的个数参数：key，Redis的namestart 字节起始位置end，字节结束位置print(r.get("foo")) # goo1 01100111 print(r.bitcount("foo",0,1)) # 11 表示前2个字节中，1出现的个数16.bitop(operation, dest, *keys)获取多个值，并将值做位运算，将最后的结果保存至新的name对应的值参数：operation,AND（并） 、 OR（或） 、 NOT（非） 、 XOR（异或）dest, 新的Redis的name*keys,要查找的Redis的name如：bitop("AND", 'new_name', 'n1', 'n2', 'n3') 获取Redis中n1,n2,n3对应的值，然后讲所有的值做位运算（求并集），然后将结果保存 new_name 对应的值中 r.set("foo","1") # 0110001 r.set("foo1","2") # 0110010 print(r.mget("foo","foo1")) # ['goo1', 'baaanew'] print(r.bitop("AND","new","foo","foo1")) # "new" 0 0110000 print(r.mget("foo","foo1","new")) source = "12" for i in source: num = ord(i) print(num) # 打印每个字母字符或者汉字字符对应的ascii码值 f-102-0b100111-01100111 print(bin(num)) # 打印每个10进制ascii码值转换成二进制的值 0b1100110（0b表示二进制） print bin(num).replace('b','') # 将二进制0b1100110替换成0110011017.strlen(name)返回name对应值的字节长度（一个汉字3个字节）print(r.strlen("foo")) # 4 'goo1'的长度是418.incr(self, name, amount=1)自增 name对应的值，当name不存在时，则创建name＝amount，否则，则自增。参数：name,Redis的nameamount,自增数（必须是整数）注：同incrbyr.set("foo", 123) print(r.mget("foo", "foo1", "foo2", "k1", "k2")) r.incr("foo", amount=1) print(r.mget("foo", "foo1", "foo2", "k1", "k2"))应用场景 – 页面点击数假定我们对一系列页面需要记录点击次数。例如论坛的每个帖子都要记录点击次数，而点击次数比回帖的次数的多得多。如果使用关系数据库来存储点击，可能存在大量的行级锁争用。所以，点击数的增加使用redis的INCR命令最好不过了。当redis服务器启动时，可以从关系数据库读入点击数的初始值（12306这个页面被访问了34634次）r.set("visit:12306:totals", 34634) print(r.get("visit:12306:totals"))每当有一个页面点击，则使用INCR增加点击数即可。r.incr("visit:12306:totals") r.incr("visit:12306:totals")页面载入的时候则可直接获取这个值print(r.get("visit:12306:totals"))19.incrbyfloat(self, name, amount=1.0)自增 name对应的值，当name不存在时，则创建name＝amount，否则，则自增。参数：name,Redis的nameamount,自增数（浮点型）r.set("foo1", "123.0") r.set("foo2", "221.0") print(r.mget("foo1", "foo2")) r.incrbyfloat("foo1", amount=2.0) r.incrbyfloat("foo2", amount=3.0) print(r.mget("foo1", "foo2"))20.decr(self, name, amount=1)自减 name对应的值，当name不存在时，则创建name＝amount，否则，则自减。参数：name,Redis的nameamount,自减数（整数)r.decr("foo4", amount=3) # 递减3 r.decr("foo1", amount=1) # 递减1 print(r.mget("foo1", "foo4"))21.append(key, value)在redis name对应的值后面追加内容参数：key, redis的namevalue, 要追加的字符串r.append("name", "haha") # 在name对应的值junxi后面追加字符串haha print(r.mget("name"))4、redis基本命令 hash1.单个增加--修改(单个取出)--没有就新增，有的话就修改hset(name, key, value)name对应的hash中设置一个键值对（不存在，则创建；否则，修改）参数：name，redis的namekey，name对应的hash中的keyvalue，name对应的hash中的value注：hsetnx(name, key, value),当name对应的hash中不存在当前key时则创建（相当于添加）import redis import time pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) r.hset("hash1", "k1", "v1") r.hset("hash1", "k2", "v2") print(r.hkeys("hash1")) # 取hash中所有的key print(r.hget("hash1", "k1")) # 单个取hash的key对应的值 print(r.hmget("hash1", "k1", "k2")) # 多个取hash的key对应的值 r.hsetnx("hash1", "k2", "v3") # 只能新建 print(r.hget("hash1", "k2"))2 批量增加（取出）hmset(name, mapping)在name对应的hash中批量设置键值对参数：name，redis的namemapping，字典，如：{'k1':'v1', 'k2': 'v2'}如：r.hmset("hash2", {"k2": "v2", "k3": "v3"})hget(name,key)在name对应的hash中获取根据key获取valuehmget(name, keys, *args)在name对应的hash中获取多个key的值参数：name，reids对应的namekeys，要获取key集合，如：['k1', 'k2', 'k3']*args，要获取的key，如：k1,k2,k3如：print(r.hget("hash2", "k2")) # 单个取出"hash2"的key-k2对应的value print(r.hmget("hash2", "k2", "k3")) # 批量取出"hash2"的key-k2 k3对应的value --方式1 print(r.hmget("hash2", ["k2", "k3"])) # 批量取出"hash2"的key-k2 k3对应的value --方式23.取出所有的键值对hgetall(name)获取name对应hash的所有键值print(r.hgetall("hash1"))4.得到所有键值对的格式 hash长度hlen(name)获取name对应的hash中键值对的个数print(r.hlen("hash1"))5.得到所有的keys（类似字典的取所有keys）hkeys(name)获取name对应的hash中所有的key的值print(r.hkeys("hash1"))6.得到所有的value（类似字典的取所有value）hvals(name)获取name对应的hash中所有的value的值print(r.hvals("hash1"))7.判断成员是否存在（类似字典的in）hexists(name, key)检查name对应的hash是否存在当前传入的keyprint(r.hexists("hash1", "k4")) # False 不存在 print(r.hexists("hash1", "k1")) # True 存在8.删除键值对hdel(name,*keys)将name对应的hash中指定key的键值对删除print(r.hgetall("hash1")) r.hset("hash1", "k2", "v222") # 修改已有的key k2 r.hset("hash1", "k11", "v1") # 新增键值对 k11 r.hdel("hash1", "k1") # 删除一个键值对 print(r.hgetall("hash1"))9.自增自减整数(将key对应的value--整数 自增1或者2，或者别的整数 负数就是自减)hincrby(name, key, amount=1)自增name对应的hash中的指定key的值，不存在则创建key=amount参数：name，redis中的namekey， hash对应的keyamount，自增数（整数）r.hset("hash1", "k3", 123) r.hincrby("hash1", "k3", amount=-1) print(r.hgetall("hash1")) r.hincrby("hash1", "k4", amount=1) # 不存在的话，value默认就是1 print(r.hgetall("hash1"))10.自增自减浮点数(将key对应的value--浮点数 自增1.0或者2.0，或者别的浮点数 负数就是自减)hincrbyfloat(name, key, amount=1.0)自增name对应的hash中的指定key的值，不存在则创建key=amount参数：name，redis中的namekey， hash对应的keyamount，自增数（浮点数）自增name对应的hash中的指定key的值，不存在则创建key=amountr.hset("hash1", "k5", "1.0") r.hincrbyfloat("hash1", "k5", amount=-1.0) # 已经存在，递减-1.0 print(r.hgetall("hash1")) r.hincrbyfloat("hash1", "k6", amount=-1.0) # 不存在，value初始值是-1.0 每次递减1.0 print(r.hgetall("hash1"))11.取值查看--分片读取hscan(name, cursor=0, match=None, count=None)增量式迭代获取，对于数据大的数据非常有用，hscan可以实现分片的获取数据，并非一次性将数据全部获取完，从而放置内存被撑爆参数：name，redis的namecursor，游标（基于游标分批取获取数据）match，匹配指定key，默认None 表示所有的keycount，每次分片最少获取个数，默认None表示采用Redis的默认分片个数如：第一次：cursor1, data1 = r.hscan('xx', cursor=0, match=None, count=None)第二次：cursor2, data1 = r.hscan('xx', cursor=cursor1, match=None, count=None)...直到返回值cursor的值为0时，表示数据已经通过分片获取完毕print(r.hscan("hash1"))12.hscan_iter(name, match=None, count=None)利用yield封装hscan创建生成器，实现分批去redis中获取数据参数：match，匹配指定key，默认None 表示所有的keycount，每次分片最少获取个数，默认None表示采用Redis的默认分片个数如：for item in r.hscan_iter('hash1'): print(item) print(r.hscan_iter("hash1")) # 生成器内存地址5、redis基本命令 list1.增加（类似于list的append，只是这里是从左边新增加）--没有就新建lpush(name,values)在name对应的list中添加元素，每个新的元素都添加到列表的最左边如：import redis import time pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) r.lpush("list1", 11, 22, 33) print(r.lrange('list1', 0, -1))保存顺序为: 33,22,11扩展：r.rpush("list2", 11, 22, 33) # 表示从右向左操作 print(r.llen("list2")) # 列表长度 print(r.lrange("list2", 0, 3)) # 切片取出值，范围是索引号0-32.增加（从右边增加）--没有就新建r.rpush("list2", 44, 55, 66) # 在列表的右边，依次添加44,55,66 print(r.llen("list2")) # 列表长度 print(r.lrange("list2", 0, -1)) # 切片取出值，范围是索引号0到-1(最后一个元素)3.往已经有的name的列表的左边添加元素，没有的话无法创建lpushx(name,value)在name对应的list中添加元素，只有name已经存在时，值添加到列表的最左边更多：r.lpushx("list10", 10) # 这里list10不存在 print(r.llen("list10")) # 0 print(r.lrange("list10", 0, -1)) # []r.lpushx("list2", 77) # 这里"list2"之前已经存在，往列表最左边添加一个元素，一次只能添加一个 print(r.llen("list2")) # 列表长度 print(r.lrange("list2", 0, -1)) # 切片取出值，范围是索引号0到-1(最后一个元素4.往已经有的name的列表的右边添加元素，没有的话无法创建r.rpushx("list2", 99) # 这里"foo_list1"之前已经存在，往列表最右边添加一个元素，一次只能添加一个 print(r.llen("list2")) # 列表长度 print(r.lrange("list2", 0, -1)) # 切片取出值，范围是索引号0到-1(最后一个元素)5.新增（固定索引号位置插入元素）linsert(name, where, refvalue, value))在name对应的列表的某一个值前或后插入一个新值参数：name，redis的namewhere，BEFORE或AFTERrefvalue，标杆值，即：在它前后插入数据value，要插入的数据r.linsert("list2", "before", "11", "00") # 往列表中左边第一个出现的元素"11"前插入元素"00" print(r.lrange("list2", 0, -1)) # 切片取出值，范围是索引号0-最后一个元素6.修改（指定索引号进行修改）r.lset(name, index, value)对name对应的list中的某一个索引位置重新赋值参数：name，redis的nameindex，list的索引位置value，要设置的值r.lset("list2", 0, -11) # 把索引号是0的元素修改成-11 print(r.lrange("list2", 0, -1))7.删除（指定值进行删除）r.lrem(name, value, num)在name对应的list中删除指定的值参数：name，redis的namevalue，要删除的值num， num=0，删除列表中所有的指定值；num=2,从前到后，删除2个； num=1,从前到后，删除左边第1个num=-2,从后向前，删除2个r.lrem("list2", "11", 1) # 将列表中左边第一次出现的"11"删除 print(r.lrange("list2", 0, -1)) r.lrem("list2", "99", -1) # 将列表中右边第一次出现的"99"删除 print(r.lrange("list2", 0, -1)) r.lrem("list2", "22", 0) # 将列表中所有的"22"删除 print(r.lrange("list2", 0, -1))8.删除并返回lpop(name)在name对应的列表的左侧获取第一个元素并在列表中移除，返回值则是第一个元素更多：rpop(name) 表示从右向左操作r.lpop("list2") # 删除列表最左边的元素，并且返回删除的元素 print(r.lrange("list2", 0, -1)) r.rpop("list2") # 删除列表最右边的元素，并且返回删除的元素 print(r.lrange("list2", 0, -1))9.删除索引之外的值ltrim(name, start, end)在name对应的列表中移除没有在start-end索引之间的值参数：name，redis的namestart，索引的起始位置end，索引结束位置r.ltrim("list2", 0, 2) # 删除索引号是0-2之外的元素，值保留索引号是0-2的元素 print(r.lrange("list2", 0, -1))10.取值（根据索引号取值）lindex(name, index)在name对应的列表中根据索引获取列表元素print(r.lindex("list2", 0)) # 取出索引号是0的值11.移动 元素从一个列表移动到另外一个列表rpoplpush(src, dst)从一个列表取出最右边的元素，同时将其添加至另一个列表的最左边参数：src，要取数据的列表的namedst，要添加数据的列表的namer.rpoplpush("list1", "list2") print(r.lrange("list2", 0, -1))12.移动 元素从一个列表移动到另外一个列表 可以设置超时brpoplpush(src, dst, timeout=0)从一个列表的右侧移除一个元素并将其添加到另一个列表的左侧参数：src，取出并要移除元素的列表对应的namedst，要插入元素的列表对应的nametimeout，当src对应的列表中没有数据时，阻塞等待其有数据的超时时间（秒），0 表示永远阻塞r.brpoplpush("list1", "list2", timeout=2) print(r.lrange("list2", 0, -1))13.一次移除多个列表blpop(keys, timeout)将多个列表排列，按照从左到右去pop对应列表的元素参数：keys，redis的name的集合timeout，超时时间，当元素所有列表的元素获取完之后，阻塞等待列表内有数据的时间（秒）, 0 表示永远阻塞 更多：r.brpop(keys, timeout) 同blpop，将多个列表排列,按照从右像左去移除各个列表内的元素r.lpush("list10", 3, 4, 5) r.lpush("list11", 3, 4, 5) while True: r.blpop(["list10", "list11"], timeout=2) print(r.lrange("list10", 0, -1), r.lrange("list11", 0, -1))14.自定义增量迭代由于redis类库中没有提供对列表元素的增量迭代，如果想要循环name对应的列表的所有元素，那么就需要：获取name对应的所有列表循环列表但是，如果列表非常大，那么就有可能在第一步时就将程序的内容撑爆，所有有必要自定义一个增量迭代的功能：def list_iter(name): """ 自定义redis列表增量迭代 :param name: redis中的name，即：迭代name对应的列表 :return: yield 返回 列表元素 """ list_count = r.llen(name) for index in range(list_count): yield r.lindex(name, index) # 使用 for item in list_iter('list2'): # 遍历这个列表 print(item)6、redis基本命令 set1.新增sadd(name,values)name对应的集合中添加元素r.sadd("set1", 33, 44, 55, 66) # 往集合中添加元素 print(r.scard("set1")) # 集合的长度是4 print(r.smembers("set1")) # 获取集合中所有的成员2.获取元素个数 类似于lenscard(name)获取name对应的集合中元素个数print(r.scard("set1")) # 集合的长度是43.获取集合中所有的成员smembers(name)获取name对应的集合的所有成员print(r.smembers("set1")) # 获取集合中所有的成员获取集合中所有的成员--元组形式sscan(name, cursor=0, match=None, count=None)print(r.sscan("set1"))获取集合中所有的成员--迭代器的方式sscan_iter(name, match=None, count=None)同字符串的操作，用于增量迭代分批获取元素，避免内存消耗太大for i in r.sscan_iter("set1"): print(i)4.差集sdiff(keys, *args)在第一个name对应的集合中且不在其他name对应的集合的元素集合r.sadd("set2", 11, 22, 33) print(r.smembers("set1")) # 获取集合中所有的成员 print(r.smembers("set2")) print(r.sdiff("set1", "set2")) # 在集合set1但是不在集合set2中 print(r.sdiff("set2", "set1")) # 在集合set2但是不在集合set1中5.差集--差集存在一个新的集合中sdiffstore(dest, keys, *args)获取第一个name对应的集合中且不在其他name对应的集合，再将其新加入到dest对应的集合中r.sdiffstore("set3", "set1", "set2") # 在集合set1但是不在集合set2中 print(r.smembers("set3")) # 获取集合3中所有的成员6.交集sinter(keys, *args)获取多一个name对应集合的交集print(r.sinter("set1", "set2")) # 取2个集合的交集7.交集--交集存在一个新的集合中sinterstore(dest, keys, *args)获取多一个name对应集合的并集，再将其加入到dest对应的集合中print(r.sinterstore("set3", "set1", "set2")) # 取2个集合的交集 print(r.smembers("set3"))并集sunion(keys, *args)获取多个name对应的集合的并集print(r.sunion("set1", "set2")) # 取2个集合的并集并集--并集存在一个新的集合sunionstore(dest,keys, *args)获取多一个name对应的集合的并集，并将结果保存到dest对应的集合中print(r.sunionstore("set3", "set1", "set2")) # 取2个集合的并集 print(r.smembers("set3"))8.判断是否是集合的成员 类似insismember(name, value)检查value是否是name对应的集合的成员，结果为True和Falseprint(r.sismember("set1", 33)) # 33是集合的成员 print(r.sismember("set1", 23)) # 23不是集合的成员9.移动smove(src, dst, value)将某个成员从一个集合中移动到另外一个集合r.smove("set1", "set2", 44) print(r.smembers("set1")) print(r.smembers("set2"))10.删除--随机删除并且返回被删除值spop(name)从集合移除一个成员，并将其返回,说明一下，集合是无序的，所有是随机删除的print(r.spop("set2")) # 这个删除的值是随机删除的，集合是无序的 print(r.smembers("set2"))11.删除--指定值删除srem(name, values)在name对应的集合中删除某些值print(r.srem("set2", 11)) # 从集合中删除指定值 11 print(r.smembers("set2"))7、redis基本命令 有序setSet操作，Set集合就是不允许重复的列表，本身是无序的有序集合，在集合的基础上，为每元素排序；元素的排序需要根据另外一个值来进行比较，所以，对于有序集合，每一个元素有两个值，即：值和分数，分数专门用来做排序。1.新增zadd(name, args, *kwargs)在name对应的有序集合中添加元素如：import redis import time pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) r.zadd("zset1", n1=11, n2=22) r.zadd("zset2", 'm1', 22, 'm2', 44) print(r.zcard("zset1")) # 集合长度 print(r.zcard("zset2")) # 集合长度 print(r.zrange("zset1", 0, -1)) # 获取有序集合中所有元素 print(r.zrange("zset2", 0, -1, withscores=True)) # 获取有序集合中所有元素和分数2.获取有序集合元素个数 类似于lenzcard(name)获取name对应的有序集合元素的数量print(r.zcard("zset1")) # 集合长度3.获取有序集合的所有元素r.zrange( name, start, end, desc=False, withscores=False, score_cast_func=float)按照索引范围获取name对应的有序集合的元素参数：name，redis的namestart，有序集合索引起始位置（非分数）end，有序集合索引结束位置（非分数）desc，排序规则，默认按照分数从小到大排序withscores，是否获取元素的分数，默认只获取元素的值score_cast_func，对分数进行数据转换的函数3-1 从大到小排序(同zrange，集合是从大到小排序的)zrevrange(name, start, end, withscores=False, score_cast_func=float)print(r.zrevrange("zset1", 0, -1)) # 只获取元素，不显示分数 print(r.zrevrange("zset1", 0, -1, withscores=True)) # 获取有序集合中所有元素和分数,分数倒序3-2 按照分数范围获取name对应的有序集合的元素zrangebyscore(name, min, max, start=None, num=None, withscores=False, score_cast_func=float)for i in range(1, 30): element = 'n' + str(i) r.zadd("zset3", element, i) print(r.zrangebyscore("zset3", 15, 25)) # # 在分数是15-25之间，取出符合条件的元素 print(r.zrangebyscore("zset3", 12, 22, withscores=True)) # 在分数是12-22之间，取出符合条件的元素（带分数）3-3 按照分数范围获取有序集合的元素并排序（默认从大到小排序）zrevrangebyscore(name, max, min, start=None, num=None, withscores=False, score_cast_func=float)print(r.zrevrangebyscore("zset3", 22, 11, withscores=True)) # 在分数是22-11之间，取出符合条件的元素 按照分数倒序3-4 获取所有元素--默认按照分数顺序排序zscan(name, cursor=0, match=None, count=None, score_cast_func=float)print(r.zscan("zset3"))3-5 获取所有元素--迭代器zscan_iter(name, match=None, count=None,score_cast_func=float)for i in r.zscan_iter("zset3"): # 遍历迭代器 print(i)4.zcount(name, min, max)获取name对应的有序集合中分数 在 [min,max] 之间的个数print(r.zrange("zset3", 0, -1, withscores=True)) print(r.zcount("zset3", 11, 22))5.自增zincrby(name, value, amount)自增name对应的有序集合的 name 对应的分数r.zincrby("zset3", "n2", amount=2) # 每次将n2的分数自增2 print(r.zrange("zset3", 0, -1, withscores=True))6.获取值的索引号zrank(name, value)获取某个值在 name对应的有序集合中的索引（从 0 开始）更多：zrevrank(name, value)，从大到小排序print(r.zrank("zset3", "n1")) # n1的索引号是0 这里按照分数顺序（从小到大） print(r.zrank("zset3", "n6")) # n6的索引号是1 print(r.zrevrank("zset3", "n1")) # n1的索引号是29 这里安照分数倒序（从大到小）7.删除--指定值删除zrem(name, values)删除name对应的有序集合中值是values的成员r.zrem("zset3", "n3") # 删除有序集合中的元素n3 删除单个 print(r.zrange("zset3", 0, -1))8.删除--根据排行范围删除，按照索引号来删除zremrangebyrank(name, min, max)根据排行范围删除r.zremrangebyrank("zset3", 0, 1) # 删除有序集合中的索引号是0, 1的元素 print(r.zrange("zset3", 0, -1))9.删除--根据分数范围删除zremrangebyscore(name, min, max)根据分数范围删除r.zremrangebyscore("zset3", 11, 22) # 删除有序集合中的分数是11-22的元素 print(r.zrange("zset3", 0, -1))10.获取值对应的分数zscore(name, value)获取name对应有序集合中 value 对应的分数print(r.zscore("zset3", "n27")) # 获取元素n27对应的分数278、其他常用操作1.删除delete(*names)根据删除redis中的任意数据类型（string、hash、list、set、有序set）r.delete("gender") # 删除key为gender的键值对2.检查名字是否存在exists(name)检测redis的name是否存在，存在就是True，False 不存在print(r.exists("zset1"))3.模糊匹配keys(pattern='*') 根据模型获取redis的name 更多： KEYS * 匹配数据库中所有 key 。 KEYS h?llo 匹配 hello ， hallo 和 hxllo 等。 KEYS h*llo 匹配 hllo 和 heeeeello 等。 KEYS h[ae]llo 匹配 hello 和 hallo ，但不匹配 hilloprint(r.keys("foo*"))4.设置超时时间expire(name ,time)为某个redis的某个name设置超时时间r.lpush("list5", 11, 22) r.expire("list5", time=3) print(r.lrange("list5", 0, -1)) time.sleep(3) print(r.lrange("list5", 0, -1))5.重命名rename(src, dst)对redis的name重命名r.lpush("list5", 11, 22) r.rename("list5", "list5-1")6.随机获取namerandomkey()随机获取一个redis的name（不删除）print(r.randomkey())7.获取类型type(name)获取name对应值的类型print(r.type("set1")) print(r.type("hash2"))8.查看所有元素scan(cursor=0, match=None, count=None)print(r.hscan("hash2")) print(r.sscan("set3")) print(r.zscan("zset2")) print(r.getrange("foo1", 0, -1)) print(r.lrange("list2", 0, -1)) print(r.smembers("set3")) print(r.zrange("zset3", 0, -1)) print(r.hgetall("hash1"))9.查看所有元素--迭代器scan_iter(match=None, count=None)for i in r.hscan_iter("hash1"): print(i) for i in r.sscan_iter("set3"): print(i) for i in r.zscan_iter("zset3"): print(i)other 方法print(r.get('name')) # 查询key为name的值 r.delete("gender") # 删除key为gender的键值对 print(r.keys()) # 查询所有的Key print(r.dbsize()) # 当前redis包含多少条数据 r.save() # 执行"检查点"操作，将数据写回磁盘。保存时阻塞 # r.flushdb() # 清空r中的所有数据管道（pipeline） redis默认在执行每次请求都会创建（连接池申请连接）和断开（归还连接池）一次连接操作，如果想要在一次请求中指定多个命令，则可以使用pipline实现一次请求指定多个命令，并且默认情况下一次pipline 是原子性操作。管道（pipeline）是redis在提供单个请求中缓冲多条服务器命令的基类的子类。它通过减少服务器-客户端之间反复的TCP数据库包，从而大大提高了执行批量命令的功能。import redis import time pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True) r = redis.Redis(connection_pool=pool) # pipe = r.pipeline(transaction=False) # 默认的情况下，管道里执行的命令可以保证执行的原子性，执行pipe = r.pipeline(transaction=False)可以禁用这一特性。 # pipe = r.pipeline(transaction=True) pipe = r.pipeline() # 创建一个管道 pipe.set('name', 'jack') pipe.set('role', 'sb') pipe.sadd('faz', 'baz') pipe.incr('num') # 如果num不存在则vaule为1，如果存在，则value自增1 pipe.execute() print(r.get("name")) print(r.get("role")) print(r.get("num"))管道的命令可以写在一起，如：pipe.set('hello', 'redis').sadd('faz', 'baz').incr('num').execute() print(r.get("name")) print(r.get("role")) print(r.get("num"))原文地址：https://www.runoob.com/w3cnote/python-redis-intro.html
记录一次开发Typecho检索图片中文字功能的实战
配置Typecho文本识别能力项目实战笔记，内容原创，转载请注明来源！前言早在去年参与苹果iOS 15内测中体验其在全局融入的文字识别能力以及相册场景文本搜索功能时，我就想给自己的博客也实现搜索图片中所含文本的功能。受限于当时的时间和技术条件，这个想法一直处于“挂起”状态。恰好上个月微信在其新版本也发布了聊天内容中的图片文本搜索功能。刚好寒假期间对PaddlePaddle深度学习框架做了些许研究并在本地部署了OCR项目。择日不如撞日，昨天晚上一口气把这套“前端”逻辑以Python脚本的形式实现了出来。程序代码已公开，可能存在不足之处与优化空间，希望各位大佬能在看过本文后不吝赐教。配置新增Typecho数据表字段：ALTER TABLE `typecho_contents` ADD `ocr` TEXT NULL AFTER `views`;修改Typecho搜索逻辑（1189行）：安装用到的Python包pip3 install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple pip3 install requests -i https://pypi.tuna.tsinghua.edu.cn/simple代码其中xxxxx应换为自己的对应信息：# -*- coding: utf-8 -*- # @Author : soarli # @Date : 2021/4/6 from datetime import datetime import os import time import requests import re import pymysql # <功能选项> # 是否更新所有信息的OCR结果(False则仅对未OCR的文章执行OCR操作) updateAll = False # 处理每张图片请求后的冷却时间(秒，防止因频次限制导致漏掉部分图片) cool = 5 # <数据库及网站配置相关信息> # 数据库地址 host = 'xxxxx' # 数据库用户名 user = 'xxxxx' # 数据库密码 password = 'xxxxx' # 数据库名 database = 'xxxxx' # 博客域名 target = 'https://xxxxx.xxxxx.xxx/' # <全局变量> # OCR接口地址 url = "https://xxxxx.xxx/xxxxx" # 全局请求头 headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.102 Safari/537.36','referer':target} upload_data = {"name": "1.jpg", "type": "image/jpeg"} # 维护数据库连接状态 def keepAlive(): cur = conn.cursor() sql = "SELECT ocr FROM `typecho_contents` WHERE cid=1" cur.execute(sql) # 输入markdown文本，输出其中引用图片中的文字结果 def getPageOCRText(text): pic_list = re.findall(r"\!\[.*\]\((.+)\)",text) this_text = '' for i in pic_list: try: img = {'file':requests.get(i,headers=headers).content} img_text = requests.post(url, upload_data, files=img) this_text += img_text.text time.sleep(cool) # 维护数据库连接状态 keepAlive() # print(this_text) except Exception as e: print('Error msg: ' + str(e)) ocrtext = this_text.replace("\n","").replace('\'','\\\'').replace('\"','\\\"').replace('-','\-') if(ocrtext == ''): ocrtext = 'none!' return ocrtext # 返回需要进行ocr操作的cid(直接传入下一个函数) def getUnOCRedCid(): cur = conn.cursor() if(updateAll): sql = "SELECT cid FROM `typecho_contents` WHERE 1" else: sql = "SELECT cid FROM `typecho_contents` WHERE ocr is NULL" try: cid_list = [] cur.execute(sql) alldata = cur.fetchall() for rec in alldata: cid_list.append(rec[0]) except Exception as e: print('Error msg: ' + str(e)) return getText(cid_list) # 获取cid列表中逐条cid对应的markdown内容并逐条调用ocr获取结果 def getText(cid_list): for i in cid_list: cur = conn.cursor() sql = "SELECT text FROM `typecho_contents` WHERE cid=%s" % (i) try: cur.execute(sql) text = cur.fetchall()[0] except Exception as e: conn.rollback() print('Error msg: ' + str(e)) # print(type(text)) print('='*50) print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())) print(i) # cid ocr_text = getPageOCRText(text[0]) # 去换行后的整篇图片OCR文本 print(ocr_text) writeOCRText(ocr_text, i) # 入库 # 将识别到的结果写入数据表的ocr字段 def writeOCRText(text, cid): cur = conn.cursor() sql = "UPDATE `typecho_contents` SET `ocr` = '%s' WHERE `typecho_contents`.`cid` = %s" % (text,cid) try: cur.execute(sql) conn.commit() except Exception as e: conn.rollback() print('Error msg: ' + str(e)) if __name__ == '__main__': print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())+'开始运行') # 连接数据库 conn = pymysql.connect(host=host, user=user, password=password, database=database) # 获取需要进行ocr操作的文章cid，其会调用getText，getText又会逐条调用getPageOCRText对markdown text引用的逐张图片进行OCR操作，返回的合成文本又会调用writeOCRText写回数据库的ocr字段 getUnOCRedCid() # 断开数据库 conn.close() print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())+'运行完毕') 执行经过本地的反复测试与修正，本程序于4月6日凌晨3点左右在我的blog服务器正式运行，并于当天晚上9点左右运行完毕。期间OCR服务器(我的另一台腾讯云)的CPU一直处于满载状态：数据看了看博客数据库中ocr字段总字符数：SELECT sum(length(ocr)) FROM `typecho_contents`我这三百多篇文章中的所有图片竟然识别出近两百万文字，真的有被吓到：效果选了6个关键词，分别截图对比了OCR程序首次执行前后的搜索结果。识别完成之前：识别完成之后：定时如果数据库中没有文章更新，程序几乎在开始运行瞬间即可运行完毕。考虑到OCR服务器的业务压力，程序的自动执行周期设置到了每天夜里。先执行一次，看看效果：完美搞定！其他不算不知道，一算吓一跳！webp格式下的我的全站图片平均每张仅占40KB的大小！Google yyds！补充04.07 4:25更新从日志里看到，这篇文章的图像识别工作已经完成：4月6号写的另一个程序，一并公开出来用途：识别任意目录下的所有图片中的文字并在该目录下生成识别结果对应的txt文件import os import requests # 填写soarli提供的OCR接口 url = "xxxx" # 填写图片目录的绝对路径 如C:/Users/xxx/Desktop/imgs filepath = 'xxxx' # 常见的图片后缀 imageexts = {'xbm', 'tif', 'pjp', 'svgz', 'jpg', 'jpeg', 'ico', 'tiff', 'gif', 'svg', 'jfif', 'webp', 'png', 'bmp', 'pjpeg', 'avif'} # 改变工作文件夹 os.chdir(filepath) # 获取总图片数 pic_num = 0 for name in os.listdir(filepath): if(name.split('.')[-1] in imageexts): pic_num = pic_num + 1 # 当前个数计数器 i = 0 # 遍历改文件夹所有的文件，并for循环 for name in os.listdir(filepath): if(name.split('.')[-1] in imageexts): i = i + 1 rate = round((i-1)/pic_num,4) print('='*50) print('正在智能识别："' + name + '"（第'+str(i)+'张，共'+str(pic_num)+'张，已完成占比'+str(rate*100)+'%）') print('='*50) img = {'file':open(name, 'rb')} result = requests.post(url, files=img) print(result.text) file = open(name+".txt", 'w', encoding='utf-8') file.write(result.text) file.close() 参考资料：https://blog.bossdong.cn/411.htmlhttps://blog.csdn.net/qq_44754132/article/details/104418657https://www.begtut.com/python/ref-requests-post.htmlhttps://blog.csdn.net/weixin_40006779/article/details/109940125https://blog.csdn.net/weixin_40394560/article/details/116269688https://blog.csdn.net/itguangzhi/article/details/107501813https://www.runoob.com/python/att-string-split.htmlhttps://blog.csdn.net/qq_38276669/article/details/81329752https://www.jianshu.com/p/801750d5a735https://blog.csdn.net/m0_37696990/article/details/105925940https://blog.csdn.net/menglinjie/article/details/78539022https://blog.csdn.net/hhyygg2009/article/details/102844607https://blog.csdn.net/weixin_50990952/article/details/111796512https://www.php.cn/python-tutorials-462304.html
前言最近发现Flask项目在运行时会报[WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead警告，意思是不要在生产环境使用这种方案启动程序：解决使用WSGI启动服务即可解决问题（中间可能需要pip3 install gevent）原版代码：from paddleocr import PaddleOCR from flask import Flask, request import cv2 import numpy app = Flask(__name__) ocr = PaddleOCR(det_model_dir='./inference/small/ch_PP-OCRv2_det_infer/', rec_model_dir='./inference/small/ch_PP-OCRv2_rec_infer/',cls_model_dir='./inference/small/ch_ppocr_mobile_v2.0_cls_infer/',use_angle_cls=True,use_space_char=True, use_gpu=False) @app.route('/xxxxxx/', methods=['get', 'POST']) def xxxxxx(): img_path = request.args.get("fname") result = ocr.ocr(img_path) ret = '' for line in result: ret += str(line[1][0]) + "\n" return str(ret) if __name__ == '__main__': app.run(host='0.0.0.0') 新版代码：from paddleocr import PaddleOCR from flask import Flask, request import cv2 import numpy from gevent import pywsgi app = Flask(__name__) ocr = PaddleOCR(det_model_dir='./inference/small/ch_PP-OCRv2_det_infer/', rec_model_dir='./inference/small/ch_PP-OCRv2_rec_infer/',cls_model_dir='./inference/small/ch_ppocr_mobile_v2.0_cls_infer/',use_angle_cls=True,use_space_char=True, use_gpu=False) @app.route('/xxxxxx/', methods=['get', 'POST']) def xxxxxx(): img_path = request.args.get("fname") result = ocr.ocr(img_path) ret = '' for line in result: ret += str(line[1][0]) + "\n" return str(ret) if __name__ == '__main__': server = pywsgi.WSGIServer(('0.0.0.0', 5000), app) server.serve_forever() 解决问题！参考资料：https://blog.csdn.net/weixin_43642491/article/details/114916769https://www.cnblogs.com/mlp1234/p/13743254.html
前言之前做过Python多线程的应用，但是一直没有系统学习，加之最近有了相关需求且发现其在Linux上的执行效率是真的高便决定简单学习并做个记录。基本用法import threading def thread_job(): print('This is an added Thread, number is %s' %threading.current_thread()) def main(): # 输出激活状态的线程数 # 1 print(threading.active_count()) # 输出当前激活状态线程的详细信息 # [<_MainThread(MainThread, started 14940)>] print(threading.enumerate()) # 输出正在运行的线程是哪一个 # <_MainThread(MainThread, started 14940)> print(threading.current_thread()) # 新建一个线程 # This is an added Thread, number is <Thread(Thread-1, started 9948)> added_thread = threading.Thread(target=thread_job) added_thread.start() if __name__ == '__main__': main()join方法默认情况下，多线程任务中子线程的运行并不会阻塞主线程，使用join方法即可对主线程进行阻塞。import threading from time import sleep def thread_job(): print('T1 start\n') for i in range(10): sleep(0.1) print('T1 finish\n') def main(): added_thread = threading.Thread(target=thread_job, name="T1") added_thread.start() print('all done\n') if __name__ == '__main__': main()输出顺序：T1 start all done T1 finish阻塞状态：import threading from time import sleep def thread_job(): print('T1 start\n') for i in range(10): sleep(0.1) print('T1 finish\n') def main(): added_thread = threading.Thread(target=thread_job, name="T1") added_thread.start() added_thread.join() # 随后的语句将被阻塞 print('all done\n') if __name__ == '__main__': main()输出顺序：T1 start T1 finish all done双线程模型：import threading from time import sleep def thread_job(): print('T1 start\n') for i in range(10): sleep(0.1) print('T1 finish\n') # 一个执行较快的函数 def T2_job(): print('T2 start\n') print('T2 finish\n') def main(): added_thread = threading.Thread(target=thread_job, name="T1") thread_2 = threading.Thread(target=T2_job, name="T2") added_thread.start() thread_2.start() added_thread.join() # 随后的语句将被阻塞 print('all done\n') if __name__ == '__main__': main()输出结果：T1 start T2 start T2 finish T1 finish all done即便是运行较快的线程，只要不阻塞，其优先级也会低于主线程：import threading from time import sleep def thread_job(): print('T1 start\n') for i in range(10): sleep(0.1) print('T1 finish\n') # 一个执行较快的函数 def T2_job(): print('T2 start\n') print('T2 finish\n') def main(): added_thread = threading.Thread(target=thread_job, name="T1") thread_2 = threading.Thread(target=T2_job, name="T2") added_thread.start() added_thread.join() # 随后的语句将被阻塞 thread_2.start() # thread_2.join() # 随后的语句将被阻塞 print('all done\n') if __name__ == '__main__': main()输出结果：T1 start T1 finish T2 start all done T2 finishQueue功能由于多线程没有返回值，我们可以将其运算结果放在一个队列中，随后每个线程的队列到主线程中后即可取出数据。以计算列表中每个元素值的平方为例：import threading from queue import Queue # 二次方的实现 def job(l,q): for i in range(len(l)): l[i] = l[i] ** 2 q.put(l) # 多线程模块 def multithreading(): # 定义队列（放返回值以取代return） q = Queue() # 所有的线程放入本列表 threads = [] data = [[1,2,3],[3,4,5],[4,4,4],[5,5,5]] # 定义4个线程 for i in range(4): # 设定线程并对每个线程传进去一个子列表和全局队列 t = threading.Thread(target=job, args=(data[i], q)) # 启动线程 t.start() # 加入到所有线程中 threads.append(t) # 阻塞每一个线程以实现异步(需要并发而不是依次执行) for thread in threads: thread.join() # 定义空列表(用来存每一个线程返回过来的值) results = [] for _ in range(4): # 按顺序从q中取出值存到results results.append(q.get()) print(results) if __name__ == '__main__': multithreading()全局解释器锁GIL未必提高效率Python的多线程，只有用于I/O密集型程序时效率才会有明显的提高。原因如下：Python代码的执行是由Python虚拟机进行控制。它在主循环中同时只能有一个控制线程在执行，意思就是Python解释器中可以运行多个线程，但是在执行的只有一个线程，其他的处于等待状态。这些线程执行是有全局解释器锁（GIL）控制，它来保证同时只有一个线程在运行。在多线程运行环境中，Python虚拟机执行方式如下：设置GIL切换进线程执行下面操作之一​ 1.运行指定数量的字节码指令​ 2.线程主动让出控制权切换出线程（线程处于睡眠状态）解锁GIL进入1步骤注意：Python运行计算密集型的多线程程序时，更倾向于让线程在整个时间片内始终占据GIL，而I/O秘籍型的多线程程序在I/O被调用前会释放GIL，以允许其他线程在I/O执行的时候运行。下面这段示例运行时多线程未必胜得过单线程：import threading import time import copy from queue import Queue def job(l, q): res = sum(l) q.put(res) def multithreading(l): q = Queue() threads = [] for i in range(4): t = threading.Thread(target=job, args=(copy.copy(l),q),name='T%i'%i) t.start() threads.append(t) [t.join() for t in threads] total = 0 for _ in range(4): total += q.get() print(total) def normal(l): total = sum(l) print(total) if __name__ == '__main__': l = list(range(1000000)) s_t=time.time() normal(l*4) print('normal:',time.time()-s_t) s_t=time.time() multithreading(l) print('multithreading:',time.time()-s_t) Lock线程锁第一个线程处理完再开始第二个线程（针对shared memory场景）：import threading def job1(): global A, lock # 打开线程锁 lock.acquire() for i in range(10): A += 1 print('job1', A) # 关闭线程锁 lock.release() def job2(): global A, lock # 打开线程锁 lock.acquire() for i in range(10): A += 10 print('job2', A) # 关闭线程锁 lock.release() if __name__ == '__main__': lock = threading.Lock() A = 0 t1 = threading.Thread(target=job1) t2 = threading.Thread(target=job2) t1.start() t2.start() t1.join() t2.join()参考资料：https://zhuanlan.zhihu.com/p/91601448https://www.cnblogs.com/guyuyun/p/11185832.htmlhttps://www.bilibili.com/video/BV1jW411Y7Wjhttps://docs.python.org/zh-cn/3/library/threading.html