Python way to explore https:https://cdn.v2ex.com/navatar/8613/985e/90_normal.png?m=1648339948 https:https://cdn.v2ex.com/navatar/8613/985e/90_large.png?m=1648339948 https://www.v2ex.com/ 2025-03-18T05:11:07Z Copyright © 2010-2018, V2EX nvim 怎么设置跳转的光标的位置 tag:www.v2ex.com,2025-03-18:/t/1119256 2025-03-18T02:52:22Z 2025-03-18T05:11:07Z kkklk https://www.v2ex.com/member/kkklk Python 写了个 GUI 程序,运行久了老是闪退 咋解决 tag:www.v2ex.com,2025-03-18:/t/1119246 2025-03-18T02:30:56Z 2025-03-18T07:20:48Z shuyouning32 https://www.v2ex.com/member/shuyouning32
gui 对象中有个 tableList 数组,数组中有个对象 执行 线程 run 任务。
gui 定时了一个任务 每 1 分钟去刷 数组对象 的 run 任务
刷久了 一晚上 起来看不到程序了。 头疼死了。

# 定时任务
def timer_task(self):
# 定时任务的具体逻辑
print("定时任务执行中...")
# 使用线程池并发执行
# with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
# futures = [executor.submit(instance.start) for instance in self.tableList]
# # 等待所有线程完成
# for future in concurrent.futures.as_completed(futures):
# future.result()
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor: # 调整线程池大小
futures = []
for instance in self.tableList:
if not self.task_queue.full(): # 检查队列是否已满
self.task_queue.put(instance)
future = executor.submit(self.safe_start, instance)
futures.append(future)
# 等待所有线程完成
for future in concurrent.futures.as_completed(futures):
future.result()
self.task_queue.get() # 任务完成后从队列中移除
# 执行任务
def safe_start(self, instance):
with self.lock: # 使用线程锁保护共享资源
try:
instance.start()
except Exception as e:
print(f"任务执行失败: {e}")
有木有大佬能帮忙看看的! ]]>
聊一个最近在做的 post-mortem debugging tool 吧 tag:www.v2ex.com,2025-03-16:/t/1118890 2025-03-16T17:07:31Z 2025-03-16T23:07:31Z gaogaotiantian https://www.v2ex.com/member/gaogaotiantian Post-mortem debugging 一直是一个很有趣的方向,它可以很大程度上解决 bug 不复现或者难复现的问题。一个很常见的例子是,在 CI 上一个 test fail 了,但是本地怎么跑都没问题,就很痛苦。logging 这个事情永远没有一个正确的量,少了信息不够,多了信息找不到。我们期待的是可以“重建”这个测试挂掉时候的状态(当然更理想的是把整个程序的运行过程重建,那是个更大的话题了)。

我在这个领域做了不少尝试,然后最近做出来一个自己还比较满意的东西,coredumpy

https://github.com/gaogaotiantian/coredumpy

这是个解决 python 的 post-mortem debugging 的工具。它的核心点其实就是一个,可以对程序当前运行状态做一个 snapshot ,然后保存 dump ,在进程结束后用这个 dump 文件去重建。

这个概念本身并不算新,在一些 native 语言里是存在的。python 这边也有人做过一些尝试,但是效果都不好。一个核心原因就是 Python 对象之间的引用关系比较复杂,不太好重建。而之前大量的尝试基本都是基于 pickle 的,pickle 有几个核心问题导致它上不了 production:

  • 最重要的,有些 object 是没法 pickle 的,而 pickle 是不支持 partial pickle 的,直接导致你项目里有一个东西没法 pickle 就 dump 不出来东西
  • pickle 有安全问题,这是 by design 的,没法避免
  • pickle 面对一些 recursion 很重的 structure 的时候可能直接爆栈
  • pickle 在还原数据的时候,是需求同样的环境的,这就减弱了“拿个 dump 直接看”的能力

所以 coredumpy 是完全抛弃了 pickle ,用另一种方式来试图解决这个问题——我可以没办法重建全部数据,但是我一定要能 dump 出来,并且保留更多的信息。

demo

拿到这个 dump 之后,可以用 vscode 直接打开(会调用 vscode 自身的 debugger ),也可以用 pdb 打开。同时还做了个 github actions ,可以在 github actions 里直接产生一个 vscode link ,然后直接复制到浏览器就可以在 vscode 打开这个 dump 了,连手动下载都不需要。

项目本身还在做,感兴趣的小伙伴可以试一下,如果有反馈也可以直接在 github 提 issue~

]]>
如何在 jupyter notebook 里安装 binance 模块?有没有会 Python 的大神来指点下? tag:www.v2ex.com,2025-03-14:/t/1118529 2025-03-14T11:53:13Z 2025-03-14T13:34:28Z flyouting https://www.v2ex.com/member/flyouting 在 jupyter notebook 里输入!pip install python-binance 然后运行,它就提示下面这堆。问了 chatgpt 也没用。有没有会 python 的大神来指点下? Defaulting to user installation because normal site-packages is not writeable Requirement already satisfied: python-binance in c:\users\mazzy\appdata\roaming\python\python312\site-packages (1.0.28) Requirement already satisfied: requests in c:\programdata\anaconda3\lib\site-packages (from python-binance) (2.32.3) Requirement already satisfied: six in c:\programdata\anaconda3\lib\site-packages (from python-binance) (1.16.0) Requirement already satisfied: dateparser in c:\users\mazzy\appdata\roaming\python\python312\site-packages (from python-binance) (1.2.1) Requirement already satisfied: aiohttp in c:\programdata\anaconda3\lib\site-packages (from python-binance) (3.10.5) Requirement already satisfied: websockets in c:\users\mazzy\appdata\roaming\python\python312\site-packages (from python-binance) (15.0.1) Requirement already satisfied: pycryptodome in c:\users\mazzy\appdata\roaming\python\python312\site-packages (from python-binance) (3.21.0) Requirement already satisfied: aiohappyeyeballs>=2.3.0 in c:\programdata\anaconda3\lib\site-packages (from aiohttp->python-binance) (2.4.0) Requirement already satisfied: aiosignal>=1.1.2 in c:\programdata\anaconda3\lib\site-packages (from aiohttp->python-binance) (1.2.0) Requirement already satisfied: attrs>=17.3.0 in c:\programdata\anaconda3\lib\site-packages (from aiohttp->python-binance) (23.1.0) Requirement already satisfied: frozenlist>=1.1.1 in c:\programdata\anaconda3\lib\site-packages (from aiohttp->python-binance) (1.4.0) Requirement already satisfied: multidict<7.0,>=4.5 in c:\programdata\anaconda3\lib\site-packages (from aiohttp->python-binance) (6.0.4) Requirement already satisfied: yarl<2.0,>=1.0 in c:\programdata\anaconda3\lib\site-packages (from aiohttp->python-binance) (1.11.0) Requirement already satisfied: python-dateutil>=2.7.0 in c:\programdata\anaconda3\lib\site-packages (from dateparser->python-binance) (2.9.0.post0) Requirement already satisfied: pytz>=2024.2 in c:\users\mazzy\appdata\roaming\python\python312\site-packages (from dateparser->python-binance) (2025.1) Requirement already satisfied: regex!=2019.02.19,!=2021.8.27,>=2015.06.24 in c:\programdata\anaconda3\lib\site-packages (from dateparser->python-binance) (2024.9.11) Requirement already satisfied: tzlocal>=0.2 in c:\users\mazzy\appdata\roaming\python\python312\site-packages (from dateparser->python-binance) (5.3.1) Requirement already satisfied: charset-normalizer<4,>=2 in c:\programdata\anaconda3\lib\site-packages (from requests->python-binance) (3.3.2) Requirement already satisfied: idna<4,>=2.5 in c:\programdata\anaconda3\lib\site-packages (from requests->python-binance) (3.7) Requirement already satisfied: urllib3<3,>=1.21.1 in c:\programdata\anaconda3\lib\site-packages (from requests->python-binance) (2.2.3) Requirement already satisfied: certifi>=2017.4.17 in c:\programdata\anaconda3\lib\site-packages (from requests->python-binance) (2025.1.31) Requirement already satisfied: tzdata in c:\programdata\anaconda3\lib\site-packages (from tzlocal>=0.2->dateparser->python-binance) (2023.3)

输入 import sys print(sys.executable),它显示正常 C:\ProgramData\anaconda3\python.exe

输入 import sys !{sys.executable} -m pip install python-binance 它又显示 Defaulting to user installation because normal site-packages is not writeable

]]>
Pydoll,取代 Selenium,一个无驱动异步网页自动化库 tag:www.v2ex.com,2025-03-13:/t/1118100 2025-03-13T04:33:28Z 2025-03-16T14:05:06Z Abnl https://www.v2ex.com/member/Abnl 有人了解这个很新鲜的 python 网页自动化库吗,看介绍很高效、很牛逼的样子。

之前尝试用 Selenium 写过一个自动预约健身房的小项目,结果一直过不了网页的反爬机制,打开网页一片空白... 不知道用这个能不能轻松搞成。

https://github.com/thalissonvs/pydoll

https://www.xugj520.cn/archives/pydoll-webdriver.html

]]>
Nodezator tag:www.v2ex.com,2025-03-12:/t/1117967 2025-03-12T13:49:58Z 2025-03-13T05:26:09Z Livid https://www.v2ex.com/member/Livid

看起来很有趣的一个 Python 库。

https://github.com/IndieSmiths/nodezator ]]>
交互式处理 1000 万个数据点的时间序列 tag:www.v2ex.com,2025-03-11:/t/1117679 2025-03-11T13:39:51Z 2025-03-12T11:27:47Z Malvo https://www.v2ex.com/member/Malvo 数据大约有 1000 万左右的数据点,一维时间序列,需要做预处理,很难用算法实现,希望用人工交互的形式处理,比如拉框选中,对选中的点进行拖动删除等,用 python 比较多,但是 python 常用的包处理这个数据量性能不行,请问还有什么好的方案,不限于 python 。

]]>
Python 小组作业,做个什么东西好呢? tag:www.v2ex.com,2025-03-11:/t/1117471 2025-03-11T02:43:29Z 2025-03-11T15:38:42Z lawrencelee https://www.v2ex.com/member/lawrencelee
大家帮忙想一想,先行谢过。 ]]>
linkedin 爬虫还能做吗 tag:www.v2ex.com,2025-03-04:/t/1115824 2025-03-04T07:57:24Z 2025-03-04T10:57:24Z gzshasha https://www.v2ex.com/member/gzshasha 最近在研究 LinkedIn 爬虫,不是 py 出身好难啊。有没有大佬接活。

]]>
初学 fleet,请问如何实现自动补完和开启错误提示? tag:www.v2ex.com,2025-03-01:/t/1115140 2025-03-01T12:51:43Z 2025-03-02T00:29:57Z ota https://www.v2ex.com/member/ota
fleet 很完美,几乎就是远程大杀器。够简洁,而且对远程服务器几乎没任何性能要求。我的 1c1g 都跑起来了。
自动检测 python 解释器,如果是 venv ,自动 activate ,而且能自动识别到 fish 。简直就是完美 ide ,因为可以 ssh 直接操作,所以也不用 vscode 那种可以新建 venv ,没必要,终端操作更便捷。

没缺点,就差自动补完和错误提示了。 ]]>
放弃 jupyter 挺痛苦的 tag:www.v2ex.com,2025-02-27:/t/1114697 2025-02-27T09:57:54Z 2025-02-28T07:41:07Z foveal https://www.v2ex.com/member/foveal 1. 更好的实验记录和版本管理
2. 得用好 AI
3. 复杂项目代码得管理的更好

现在感觉
1. 项目复杂了,层次多了 jupyter 不好管理
2. jupyter AI 编程支持的不那么顺手
3. 代码的版本管理不好做,想做一套升级的实验环境不如脚本好控制
4. vscode 生态发展的很好,还有了 positron 这种中间态

决定现转到 positron 上,感觉过程好痛苦,各种不适应效率降低。
习惯了开 cell 做小分析,习惯了随时跑上实验,不需要把中间变量逐个保存到本地,第二天来页面分析就能看到所有中间变量。vscode 好像不行,需要把中间结果保存到本地,否则连接断开了就没了,每次再开 screen ipython 之类的更麻烦。

有大神有更好的环境方案么? ]]>
pycharm 的高亮显示是不是有问题 tag:www.v2ex.com,2025-02-26:/t/1114287 2025-02-26T02:53:36Z 2025-02-26T09:33:39Z gegeligegeligo https://www.v2ex.com/member/gegeligegeligo 方法名的颜色会和类名的颜色混起来,
变量名的颜色会和方法名的颜色混起来。
换了好几个版本的 pycharm ,发现都是一样的
放张图:
图片

看到了吧,这个 QTableWidgetItem ,前两个是绿色,后两个是黄色,已经确认了不是主题的问题,换什么主题都会存在这样的问题。有遇到同样问题的吗?谢谢

]]>
有没有爬虫管理开源项目,可以管理爬虫脚本、ip 代理池、数据可视化 tag:www.v2ex.com,2025-02-26:/t/1114256 2025-02-26T01:54:34Z 2025-02-26T09:46:25Z nunet https://www.v2ex.com/member/nunet 有时候写了一些爬虫脚本,但都是单文件的,每次都要指定好数据库字段,然后运行爬虫爬取数据。 我就在想,有没有可以管理爬虫脚本的项目,支持将爬虫脚本编辑和运行,支持预览爬取的数据,可以配置 ip 代理池。我在 github 上找了一下,没有找到。自己开发的话,感觉是个大工程。

]]>
使用 Nuitka 默认参数打包出来的 exe 仍可以正常输出带路径、行号的错误堆栈,是否说明可以反编译?为什么? Nuitka 存储了怎样的对应关系,是不是像 il2cpp 的 global-metadata 一样 tag:www.v2ex.com,2025-02-24:/t/1113975 2025-02-24T20:13:36Z 2025-02-25T02:18:20Z drymonfidelia https://www.v2ex.com/member/drymonfidelia 感谢:请各位大佬推荐一个 Python 开源项目做毕设❤️ tag:www.v2ex.com,2025-02-23:/t/1113586 2025-02-23T05:24:17Z 2025-02-23T15:06:03Z uCharles https://www.v2ex.com/member/uCharles 如题, 朋友大学毕业想让我给推荐个 python 开源项目做毕设,但是我平常只接触 Java 啥的,难搞啊🤣 请求各位大佬给介绍几个 python 项目, 万分感谢

]]>
AIGC 应用程序开发者(远程兼职) tag:www.v2ex.com,2025-02-23:/t/1113582 2025-02-23T05:12:39Z 2025-02-24T06:01:34Z valor https://www.v2ex.com/member/valor 职位信息 职位名称:AIGC 应用程序开发者(兼职) 工作模式:远程工作,按项目合作。根据项目需求确定开发任务与报酬

岗位职责 1.负责 Coze 、LangChain 等 AIGC Framework 的工作流搭建,优化 AI 应用的开发流程。 2.熟练使用 主流大语言模型( LLM )(如 DS 、OpenAI GPT-4 、Claude 、Gemini 、Mistral 、Llama 2 等),并进行 API 集成与优化。 3.负责应用架构设计与开发,确保 AI 生成应用的高效运行与稳定性。 4.参与 AIGC 解决方案的技术选型与落地,推动 AI 生成应用的创新与优化。 5.通过远程协作,与团队成员沟通项目进展,确保按时交付高质量成果。

任职要求 1.熟悉 AIGC 开发框架(如 Coze 、LangChain 、LlamaIndex 、Haystack ),能够搭建 LLM 工作流。 2.熟练使用主流 LLM ,具备 API 调用、微调( Fine-tuning )、Prompt 设计等实践经验。 3.有丰富的后端开发经验,熟练掌握 Python ( FastAPI / Flask )或 Node.js ( Express / NestJS ) 。 4.熟悉数据库与缓存优化(如 PostgreSQL / MySQL / Redis / MongoDB ),具备良好的架构设计能力。 5.熟练使用 Docker 、Kubernetes 、Cloud 服务( AWS/GCP/Azure ) 进行部署与优化。 6.具备良好的远程协作与自我管理能力,能够独立完成任务并高效沟通。

加分项 1.有完整的 AIGC 应用开发经验,如 AI 客服、助手、AI 生成内容平台、智能对话系统等。 2.了解私有化 LLM 部署,能优化 AI 模型的推理速度与成本。 3.具备分布式系统开发经验,能够优化 AIGC 应用的高并发处理能力,熟悉本地化 AI 应用方向。

邮箱: axesong@outlook.com

]]>
django 网站接入微信登录,在手机非微信浏览器端登录,想用微信开放平台的 h5 唤起微信客户端登录,代码如下,怎么还是总提示:请在微信客户端打开链接呢? tag:www.v2ex.com,2025-02-23:/t/1113563 2025-02-23T02:26:20Z 2025-02-23T03:47:11Z python30 https://www.v2ex.com/member/python30 django 网站接入微信登录,在手机非微信浏览器端登录,想用微信开放平台的 h5 唤起微信客户端登录,代码如下,怎么还是总提示:请在微信客户端打开链接呢?

views.py

def generate_state():
    """
    生成一个随机的 state 参数
    """
    return secrets.token_urlsafe(16)  # 生成一个 16 字节的随机字符串

def wechat_login(request):
    """
    微信登录入口,根据设备类型选择不同的授权方式
    """
    user_agent = request.META.get('HTTP_USER_AGENT', '').lower()

    # 判断是否为微信浏览器
    is_wechat_browser = 'micromessenger' in user_agent

    # 判断是否为手机端
    is_mobile = any(keyword in user_agent for keyword in ['mobile', 'android', 'iphone', 'ipod'])

    # 生成随机的 state 参数
    state = generate_state()
    request.session['wechat_state'] = state  # 将 state 存储在会话中


    if is_wechat_browser:
        # 微信浏览器内登录(使用微信公众号的 OAuth2.0 授权)
        wechat_auth_url = (
            f"https://open.weixin.qq.com/connect/oauth2/authorize"
            f"?appid={settings.WECHAT_MP_APP_ID}"  # 使用微信公众号的 AppID
            f"&redirect_uri={settings.WECHAT_MP_REDIRECT_URI}"  # 微信公众号的回调地址
            f"&response_type=code"
            f"&scope=snsapi_userinfo"  # 使用 snsapi_userinfo 或 snsapi_base
            f"&state={state}#wechat_redirect"
        )
    elif is_mobile:
        # 手机端浏览器登录(使用微信开放平台的 H5 登录)
        wechat_auth_url = (
            f"https://open.weixin.qq.com/connect/oauth2/authorize"
            f"?appid={settings.WECHAT_OPEN_APP_ID}"  # 使用微信开放平台的 AppID
            f"&redirect_uri={settings.WECHAT_OPEN_REDIRECT_URI}"  # 微信开放平台的回调地址
            f"&response_type=code"
            f"&scope=snsapi_userinfo"  # 使用 snsapi_login
            f"&state={state}#wechat_redirect"
        )
    else:
        # 电脑端浏览器登录(使用微信开放平台的扫码登录)
        wechat_auth_url = (
            f"https://open.weixin.qq.com/connect/qrconnect"
            f"?appid={settings.WECHAT_OPEN_APP_ID}"  # 使用微信开放平台的 AppID
            f"&redirect_uri={settings.WECHAT_OPEN_REDIRECT_URI}"  # 微信开放平台的回调地址
            f"&response_type=code"
            f"&scope=snsapi_login"  # 使用 snsapi_login
            f"&state={state}#wechat_redirect"
        )

    return redirect(wechat_auth_url)


def get_wechat_h5_login_url(request):
    """
    生成微信开放平台的 H5 登录 URL
    """
	# 生成随机的 state 参数
    state = generate_state()
    request.session['wechat_state'] = state
    # 微信开放平台的 H5 登录 URL
    wechat_h5_login_url = (
        f"https://open.weixin.qq.com/connect/oauth2/authorize"
        f"?appid={settings.WECHAT_OPEN_APP_ID}"  # 微信开放平台的 AppID
        f"&redirect_uri={settings.WECHAT_OPEN_REDIRECT_URI}"  # 回调地址
        f"&response_type=code"
        f"&scope=snsapi_login"  # 使用 snsapi_login
        f"&state={state}"  # 可选参数,用于防止 CSRF 攻击
        f"#wechat_redirect"
    )

    return JsonResponse({
        'success': True,
        'login_url': wechat_h5_login_url,
    })


def filter_username(nickname):
    """
    过滤掉不允许的字符,只保留字母、数字、_、-
    """
    # 只保留字母、数字、_、-
    filtered = re.sub(r'[^\w-]', '', nickname)
    return filtered

def generate_unique_username(nickname):
    """
    根据微信昵称生成唯一的用户名
    """
    # 过滤特殊字符
    base_username = filter_username(nickname)
    
    # 如果过滤后的用户名为空,使用默认用户名
    if not base_username:
        base_username = 'wechat_user'
    
    # 添加随机后缀
    random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=4))
    username = f"{base_username}_{random_suffix}"
    
    # 检查用户名是否已存在
    while User.objects.filter(username=username).exists():
        random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=4))
        username = f"{base_username}_{random_suffix}"
    
    return username

 
def wechat_callback(request):
    """
    微信登录回调处理
    """
    code = request.GET.get('code')
    state = request.GET.get('state')
    if not code:
        return HttpResponse('授权失败,未获取到 code')

    # 验证 state 参数
    if state != request.session.get('wechat_state'):
        return HttpResponse('非法请求,state 参数不匹配')
	
    # 清除会话中的 state
    request.session.pop('wechat_state', None)


    # 判断是否为微信浏览器
    user_agent = request.META.get('HTTP_USER_AGENT', '').lower()
    is_wechat_browser = 'micromessenger' in user_agent

    # 根据设备类型选择 appid 和 secret
    if is_wechat_browser:
        # 微信浏览器内登录(使用微信公众号的 appid 和 secret )
        appid = settings.WECHAT_MP_APP_ID
        secret = settings.WECHAT_MP_APP_SECRET
    else:
        # 非微信浏览器登录(使用微信开放平台的 appid 和 secret )
        appid = settings.WECHAT_OPEN_APP_ID
        secret = settings.WECHAT_OPEN_APP_SECRET

    # 通过 code 获取 access_token
    token_url = (
        f"https://api.weixin.qq.com/sns/oauth2/access_token"
        f"?appid={appid}"
        f"&secret={secret}"
        f"&code={code}"
        f"&grant_type=authorization_code"
    )
    response = requests.get(token_url)
    data = response.json()

    access_token = data.get('access_token')
    openid = data.get('openid')
    unionid = data.get('unionid')  # 获取 unionid

    if not access_token or not openid:
        return HttpResponse('获取 access_token 失败')

    # 获取用户信息
    user_info_url = (
        f"https://api.weixin.qq.com/sns/userinfo"
        f"?access_token={access_token}"
        f"&openid={openid}"
    )
    response = requests.get(user_info_url)
    response.encoding = 'utf-8'
    user_info = response.json()

    # 根据 unionid 查找或创建用户
    user, created = User.objects.get_or_create(unionid=unionid)
    if created:
        # 生成唯一的用户名
        nickname = user_info.get('nickname', '微信用户')
        user.username = generate_unique_username(nickname)
        user.set_unusable_password()  # 微信登录用户不需要密码
    user.nickname = user_info.get('nickname', '')  # 更新昵称
    user.avatar = user_info.get('headimgurl', '')  # 更新头像
    user.wechat_nickname = user_info.get('nickname', '')  # 更新微信昵称
    user.save()

    # 登录用户
    login(request, user)

    return redirect('/')  # 登录成功后跳转到首页


def get_js_sdk_config(request):
    """
    获取 JS-SDK 配置
    """
    try:
        access_token = get_access_token(settings.WECHAT_MP_APP_ID, settings.WECHAT_MP_APP_SECRET)
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=500)

    # 获取 JSAPI Ticket
    ticket_url = f"https://api.weixin.qq.com/cgi-bin/ticket/getticket?access_token={access_token}&type=jsapi"
    response = requests.get(ticket_url)
    jsapi_ticket = response.json().get('ticket')

    # 生成签名参数
    nonce_str = generate_nonce_str()
    timestamp = int(time.time())
    url = request.build_absolute_uri()

    # 生成签名
    signature = generate_signature(jsapi_ticket, nonce_str, timestamp, url)

    # 返回 JS-SDK 配置
    return JsonResponse({
        'appId': settings.WECHAT_MP_APP_ID,  # 公众号的 AppID
        'timestamp': timestamp,
        'nonceStr': nonce_str,
        'signature': signature,
        'wechatOpenAppId': settings.WECHAT_OPEN_APP_ID,  # 微信开放平台的 AppID
        'wechatOpenRedirectUri': settings.WECHAT_OPEN_REDIRECT_URI,  # 微信开放平台回调地址
		'wechatMpRedirectUri': settings.WECHAT_MP_REDIRECT_URI  # 微信开放平台回调地址
    })

def get_access_token(appid, appsecret):
    """
    获取微信 access_token ,并缓存
    """
    # 先从缓存中获取 access_token
    access_token = cache.get('wechat_access_token')
    if access_token:
        return access_token

    # 缓存中没有,则从微信接口获取
    url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={appid}&secret={appsecret}"
    response = requests.get(url)
    data = response.json()
    
    if 'access_token' in data:
        access_token = data['access_token']
        # 将 access_token 缓存 7000 秒(微信的有效期是 7200 秒)
        cache.set('wechat_access_token', access_token, 7000)
        return access_token
    else:
        raise Exception(f"获取 access_token 失败: {data}")

urls.py

		path('wechat/login/', wechat_login, name='wechat_login'),
		path('wechat/callback/', wechat_callback, name='wechat_callback'),
		path('get_js_sdk_config/', get_js_sdk_config, name='get_js_sdk_config'),  # 获取 JS-SDK 配置
        path('wechat/get_wechat_h5_login_url/', get_wechat_h5_login_url, name='get_wechat_h5_login_url'),#手机端非微信浏览器登录

login.html 方式 1

<script src="/load/view.php?a=aHR0cHM6Ly9yZXMud3gucXEuY29tL29wZW4vanMvandlaXhpbi0xLjYuMC5qcw"></script>
<div class="container reg">
    <!-- 微信登录按钮 -->
    <button id="wechat-login">微信登录</button>
</div>
<script>
    document.getElementById('wechat-login').addEventListener('click', function () {
        // 判断是否为微信浏览器
        var userAgent = navigator.userAgent.toLowerCase();
        var isWechatBrowser = userAgent.indexOf('micromessenger') !== -1;
        var isMobile = /mobile|android|iphone/i.test(userAgent);

        if (isWechatBrowser || !isMobile) {
            // 微信浏览器内或电脑端,直接跳转到后端生成的微信登录 URL
            window.location.href = "/user/wechat/login/"; // 这里假设 Django 反向解析后的实际 URL
        } else {
            // 手机端非微信浏览器,使用微信开放平台的 H5 登录
            // 向后端请求微信开放平台的 H5 登录 URL
            var xhr = new XMLHttpRequest();
            xhr.open('GET', '/user/wechat/get_wechat_h5_login_url/', true); // 后端生成 H5 登录 URL 的接口
            xhr.onreadystatechange = function () {
                if (xhr.readyState === 4 && xhr.status === 200) {
                    var response = JSON.parse(xhr.responseText);
                    if (response.success) {
                        // 跳转到微信开放平台的 H5 登录页面
                        window.location.href = response.login_url;
                    } else {
                        alert('获取微信登录链接失败,请稍后重试。');
                    }
                }
            };
            xhr.send();
        }
    });
</script>

login.html 方式 2

<script src="/load/view.php?a=aHR0cHM6Ly9yZXMud3gucXEuY29tL29wZW4vanMvandlaXhpbi0xLjYuMC5qcw"></script>
<div class="container reg">
    <!-- 微信登录按钮 -->
    <button id="wechat-login">微信登录</button>
</div>
<script>
        document.getElementById('wechat-login').addEventListener('click', function () {
            // 判断是否为微信浏览器
            var userAgent = navigator.userAgent.toLowerCase();
            var isWechatBrowser = userAgent.indexOf('micromessenger')!== -1;
            var isMobile = /mobile|android|iphone/i.test(userAgent);

            if (isWechatBrowser ||!isMobile) {
                // 微信浏览器内或电脑端,直接跳转到后端生成的微信登录 URL
                window.location.href = "/user/wechat/login/"; // 这里假设 Django 反向解析后的实际 URL
            } else {
                // 获取 JS-SDK 配置
                var xhr = new XMLHttpRequest();
                xhr.open('GET', '/user/get_js_sdk_config/', true);
                xhr.onreadystatechange = function () {
                    if (xhr.readyState === 4 && xhr.status === 200) {
                        var config = JSON.parse(xhr.responseText);
                        wx.config({
                            debug: true, // 调试模式
                            appId: config.appId, // 公众号的 AppID
                            timestamp: config.timestamp, // 时间戳
                            nonceStr: config.nonceStr, // 随机字符串
                            signature: config.signature, // 签名
                            jsApiList: ['launchApplication'] // 需要使用的 JS 接口
                        });

                        wx.ready(function () {
                            // 尝试唤醒微信客户端
                            wx.launchApplication({
                                appId: config.appId, // 这里需要替换为实际的微信 AppID
                                extraData: '',
                                success: function () {
                                    // 唤醒成功,跳转到微信登录 URL
                                    window.location.href = "/user/wechat_login/"; // 这里假设 Django 反向解析后的实际 URL
                                },
                                fail: function (res) {
                                    alert('唤醒微信客户端失败,请检查您的微信是否安装或尝试其他登录方式。');
                                }
                            });
                        });

                        wx.error(function (res) {
                            alert('JS-SDK 配置出错,请刷新页面重试。');
                        });
                    }
                };
                xhr.send();
            }
        });
    </script>

setting.py 里面也配置好了上面的 appid 相关参数与回调地址,并且在电脑 pc 端,微信浏览器内都可以成功登录了,就是在手机非浏览器端,不管怎么操作,都是提示:请在微信客户端内打开链接

麻烦各位大佬看看是哪里的原因? 一直无法用 h5 的 js-sdk 在手机端唤起微信客户端登录?谢谢

]]>
请教一下各位佬, Pyqt5 中 QComboBox 选择框的文本怎么居中 tag:www.v2ex.com,2025-02-21:/t/1113175 2025-02-21T03:31:37Z 2025-02-11T03:29:37Z Henrysun https://www.v2ex.com/member/Henrysun 求助: Python poetry install Resolving dependencies 永远很慢, 如何解决 tag:www.v2ex.com,2025-02-20:/t/1113002 2025-02-20T09:02:46Z 2025-02-20T11:00:33Z Oni0n https://www.v2ex.com/member/Oni0n python poetry install Resolving dependencies 永远很慢, 如何解决

]]>
有人喜欢写爬虫用浏览器扩展来实现的吗? tag:www.v2ex.com,2025-02-20:/t/1112944 2025-02-20T07:06:42Z 2025-02-21T20:10:58Z s2555 https://www.v2ex.com/member/s2555 要爬网页的时候,每次用 python 写都是容易跳出验证,有些要登录的更麻烦,后来我发现写浏览器扩展,直接操作当前的页面好像不用考虑那么多情况,爬到的数据直接发到接口处理。所以问下大家,我这个是 Python 用得不好,技能点歪了吗?

]]>
有人在生产环境用 Sanic 吗?有什么需要注意的坑吗?看起来性能很强还轻量。我记得以前这个框架有乱关闭长连接的 Bug,花了一年多才修复 tag:www.v2ex.com,2025-02-15:/t/1111716 2025-02-15T21:35:03Z 2025-02-16T07:46:38Z drymonfidelia https://www.v2ex.com/member/drymonfidelia 用纯 Python 打造你的第一个 Web 应用 tag:www.v2ex.com,2025-02-15:/t/1111607 2025-02-15T01:48:57Z 2025-02-16T04:11:31Z daya0576 https://www.v2ex.com/member/daya0576 最近使用纯 Python 开发了一个没有目标的习惯追踪应用 (支持自部署):

优雅的状态管理

整体开发的体验非常丝滑,NiceGUI 框架提供了丰富的前端组件。

同时支持非常自然地将前端状态绑定至 python 的属性或变量:

demo = Demo()
ui.slider(min=1, max=3).bind_value(demo, 'number')

甚至直接绑定至用户一次 session:

ui.textarea('This note is kept between visits')
   .classes('w-full').bind_value(app.storage.user, 'note')

鱼与熊掌不可兼得

优雅的背后是 tradeoff ,就如同 GIL 一刀切一般,在 NiceGUI 框架中所有的逻辑都在后端服务器处理(美名称之为 backend-first philosophy )。

举个例子,在前端界面点击菜单下拉框,也需要一来一回的 websocket 交互。如果网络状况一般,可能会严重影响用户体验,这也是为什么个人项目主打自部署~

总而言之

总而言之,NiceGUI 对于没有丰富前端经验的 Python 程序员,应该会是一个不错的选择 :)

]]>
vscode Ruff 有时不格式化 f-string 内的代码什么情况? tag:www.v2ex.com,2025-02-14:/t/1111436 2025-02-14T04:34:57Z 2025-02-14T04:34:57Z wencan https://www.v2ex.com/member/wencan 两台电脑 linux 发行版不同 都安装着 vscode+Ruff 同样的项目代码 vscode 版本,一个是 1.97.2 ,一个 1.97.1 1.97.1 正常格式化 f-string ,但 1.97.2 只格式化 f-string 之外的代码,手动执行也不行 尝试了把 1.97.2 的用户数据配置文件全删了,仅安装 python 和 Ruff 两个扩展(含自动安装的),复制另一台电脑的 vscode 配置文件,但还是不能补全

]]>
使用黑魔法的异步 Python HTTP 客户端(Dev), curl-cffi 的替代品 tag:www.v2ex.com,2025-02-14:/t/1111429 2025-02-14T04:01:56Z 2025-02-15T00:21:46Z 0x676e67 https://www.v2ex.com/member/0x676e67 rnet

Powered by FFI from rquest.

Features

  • Plain, JSON, urlencoded
  • Header Order
  • Redirect Policy
  • Cookie Store
  • HTTP Proxies
  • HTTPS via BoringSSL
  • Perfectly Chrome, Safari, and Firefox

repo: https://github.com/0x676e67/rnet

]]>
Python 3.14 采用新型解释器,速度提高-3%~30% tag:www.v2ex.com,2025-02-12:/t/1111032 2025-02-12T12:11:34Z 2025-02-13T18:00:53Z codists https://www.v2ex.com/member/codists Python 官方文档( https://docs.python.org/3.14/whatsnew/3.14.html#a-new-type-of-interpreter) :

A new type of interpreter based on tail calls has been added to CPython. For certain newer compilers, this interpreter provides significantly better performance. Preliminary numbers on our machines suggest anywhere from -3% to 30% faster Python code, and a geometric mean of 9-15% faster on pyperformance depending on platform and architecture. The baseline is Python 3.14 built with Clang 19 without this new interpreter.

其实我很想知道,这种性能测试的结果到底准不准?熟悉这方面的大佬来聊聊。

]]>
有没有小白极速入门 Python 的方法 tag:www.v2ex.com,2025-02-06:/t/1109280 2025-02-06T03:24:56Z 2025-02-14T06:01:35Z LaLy https://www.v2ex.com/member/LaLy 纯小白系列

]]>
conda 中,如何兼容使用低版本 openssl 的.so 动态库? tag:www.v2ex.com,2025-02-06:/t/1109248 2025-02-06T01:50:27Z 2025-02-06T11:21:10Z nowheremanx https://www.v2ex.com/member/nowheremanx funcs.so 给 python 用,通过 ctypes 来交互。这个 otherlib 有点古老,需要低版本的 libcrypto.so ,不然有些 symbol 找不到( init 啥的函数)。我电脑里有已经编译好的 libcrypto.so.1.0.0 (在/opt/mypath/lib 下),所以编译的时候,要提供-L 和-Wl,rpath 这些参数给 gcc ,不然 funcs.so 的时候没办法找到这个版本的动态库。

折腾了很久,把这些编译的过程都写在 setup.py 中,本地跑已经没问题,用户直接把包拷过去 pip install.也没问题。 问题出现在 conda package 打包。


首先,我不能要求把 openssl 的依赖固定到低版本:虽然这个 otherlib 用了老版本的 libcrypto.so ( 1.0.0 或者 2.x.x ),但 mylib 中其他的依赖都已经用 3.x.x 。 其次,我设置的 rpath 参数不在 conda 的 build 环境中,一直给我报错,试了各种 whitelist 也没用。 这个库自带了 libressl ,我想着让 conda 直接编译了放到 env 里,但是 CONDA_PREFIX 这种东西指向的是 build 的环境,傻傻搞不清楚。

说实话,搞不懂 conda build 到底是个什么概念,这些.so 文件的编译不应该是用户那里跑的吗,为什么在我打包的时候要跑一次,那赋值 rpath 就变得非常困难了。有一个 @ORIGIN 的符号,但是这个符号本身代表的是 python 这个程序?那难道我要把东西都编译到 @ORIGIN/../lib/mylib/?


拓展一下,如果我要写一个 C++程序也用到这个 otherlib 库,但同时又有一个依赖需要高版本 libcrypto.so ,那我是不是就没办法编译了?


这个问题可能是相当深入了,不知道有没有大神能讲清楚,谢谢!!! ]]>
pycharm 如何做远程开发? tag:www.v2ex.com,2025-02-02:/t/1108670 2025-02-02T21:32:47Z 2025-02-03T20:59:04Z ota https://www.v2ex.com/member/ota 这样可以实时进行生产环境上的程序调试。

还有最大理由就是不想在本地装 py 。 ]]>
[蛇年进阶 Python !] 过年蹲家里翻译了本书《Cosmic Python 》,楼主认为是 Python 届的 DDD(领域驱动设计)手把手喂饭教学 tag:www.v2ex.com,2025-02-01:/t/1108524 2025-02-01T08:59:18Z 2025-02-01T15:42:01Z fushall https://www.v2ex.com/member/fushall 如题,

楼主从业开始用 Python 七八年了,这几年也看到了行业的不景气,

自己这几年一直很努力学习 Python 技术+领域驱动设计+需求分析,

以前的梦想就是当一个合格的 Python 软件工程师,可现实吧......哎.....

不知道自己能在这个行业还能多久,过年期间突然想着给心爱的 Python 留点儿什么,

于是开始了翻译(其实英文原文也很通俗易懂,楼主翻译完全是自己太闲了)

基本上是 AI+人工复制粘贴,这几天正在人工核对一些翻译不准确的地方...

仓库的 Readme.md 里可以看所有的文字,也没有什么商业目的,单纯希望 Python 越来越好,

这是翻译后的 github: https://github.com/fushall/cosmicpython-book

这是原作者的 github: https://github.com/cosmicpython/book

下面书的目录:

Chapter
章节
Preface
前言(已翻译)
Introduction: Why do our designs go wrong?
引言:为什么我们的设计会出问题?(已翻译)
Part 1 Intro
第一部分简介(已翻译)
Chapter 1: Domain Model
第一章:领域模型(已翻译)
Chapter 2: Repository
第二章:仓储(已翻译)
Chapter 3: Interlude: Abstractions
第三章:插曲:抽象(已翻译)
Chapter 4: Service Layer (and Flask API)
第四章:服务层(和 Flask API )(已翻译)
Chapter 5: TDD in High Gear and Low Gear
第五章:高速档与低速档中的测试驱动开发( TDD )(已翻译)
Chapter 6: Unit of Work
第六章:工作单元(已翻译)
Chapter 7: Aggregates
第七章:聚合(已翻译)
Part 2 Intro
第二部分简介(已翻译)
Chapter 8: Domain Events and a Simple Message Bus
第八章:领域事件与简单消息总线(已翻译)
Chapter 9: Going to Town on the MessageBus
第九章:深入探讨消息总线(已翻译)
Chapter 10: Commands
第十章:命令(已翻译)
Chapter 11: External Events for Integration
第十一章:集成外部事件(已翻译)
Chapter 12: CQRS
第十二章:命令查询责任分离(已翻译)
Chapter 13: Dependency Injection
第十三章:依赖注入(已翻译)
Epilogue: How do I get there from here?
尾声:我该如何开始?(已翻译)
Appendix A: Recap table
附录 A:总结表格(已翻译)
Appendix B: Project Structure
附录 B:项目结构(已翻译)
Appendix C: A major infrastructure change, made easy
附录 C:轻松替换重要的基础设施(已翻译)
Appendix D: Django
附录 D:Django (已翻译)
Appendix F: Validation
附录 F:校验(已翻译)
]]>
cf 保护下的 json 采集 tag:www.v2ex.com,2025-02-01:/t/1108488 2025-02-01T02:35:11Z 2025-02-01T03:56:19Z lsww https://www.v2ex.com/member/lsww scrapy 的 item 队列把内存挤爆 tag:www.v2ex.com,2025-01-29:/t/1108282 2025-01-29T10:45:40Z 2025-02-06T07:56:58Z bwijn https://www.v2ex.com/member/bwijn scrapy 的 spider 抓取了很多 item 然后交给 pipeline, 但是 pipeline 处理的太慢了,我的业务很耗时,但是 spider 还在不停抓取,导致任务积压

我目前打算使用 arq(异步消息框架) spider 抓到的 item 直接扔给 arq,扔之前查看任务队列是否过多,可以暂停发布任务

各位有好的想法吗

]]>
初一的孩子想要学 Python , 有好的视频集合推荐吗, 孩子能看得懂的书也可以 tag:www.v2ex.com,2025-01-27:/t/1108115 2025-01-27T13:50:56Z 2025-01-28T12:51:19Z admirez https://www.v2ex.com/member/admirez pyinstaller 打包的奇怪问题 tag:www.v2ex.com,2025-01-26:/t/1107916 2025-01-26T03:53:21Z 2025-01-31T07:56:11Z balabalaXMX https://www.v2ex.com/member/balabalaXMX 我用 pyexiv2 和 pyqt5 写了一个很简单的 demo ,想试一下这个打包,但是我发现在 macbook 里很容易就打出来了,把 pyexiv2 依赖的动态库 exiv2.dll 和 lib/exiv2api.dylib 加载进去就可以了,但是在 windows 下却有问题。

问题并不是打包失败或者运行报错,而是有时候可以有时候不行。不行的时候也不报错,而是就是好像跳过那一行直接返回了。

]]>
《 Python 工匠》中多态的使用有一些疑惑 tag:www.v2ex.com,2025-01-24:/t/1107575 2025-01-24T05:39:55Z 2025-01-24T20:11:47Z shinonome https://www.v2ex.com/member/shinonome 文中给了一个案例

class FancyLogger:
    """日志类:支持向文件、Redis 、ES 等服务输出日志"""

    _redis_max_length = 1024

    def __init__(self, output_type=OutputType.FILE):
        self.output_type = output_type
        ...

    def log(self, message):
        """打印日志"""
        if self.output_type == OutputType.FILE:
            ...
        elif self.output_type == OutputType.REDIS:
            ...
        elif self.output_type == OutputType.ES:
            ...
        else:
            raise TypeError('output type invalid')

    def pre_process(self, message):
        """预处理日志"""
        # Redis 对日志最大长度有限制,需要进行裁剪
        if self.output_type == OutputType.REDIS:
            return message[:self._redis_max_length]

FancyLogger 类在日志输出类型不同时,需要有不同的行为。因此,我们完全可以为“输出日志”行为建模一个新的类型:LogWriter ,然后把每个类型的不同逻辑封装到各自的 Writer 类中。

class FileWriter:
    def write(self, message):
        ...
        

class RedisWriter:
    max_length = 1024

    def write(self, message):
        message = self._pre_process(message)
        ...

    def _pre_process(self, message):
        """Redis 对日志最大长度有限制,需要进行裁剪"""
        return message[:self.max_length]


class EsWriter:
    def write(self, message):
        ...

基于这些不同的 Writer 类,FancyLogger 可以简化成下面这样:

class FancyLogger:
    """日志类:支持向文件、Redis 、ES 等服务输出日志"""

    def __init__(self, output_writer=None):
        self._writer = output_writer or FileWriter()
        ...

    def log(self, message):
        self._writer.write(message)

文中对这样的写法好处解释为 代码利用多态特性,完全消除了原来的条件判断语句。另外你会发现,新代码的扩展性也远比旧代码好。

但是在我看来, 你要传什么 output_writer 不还是要通过 if 来选择吗, 只是把一个地方的 if 换到了另外一个地方,

扩展性 这个模块看起来确实好了, 但是总感觉和上面一样, 这里提高了, 但是其他地方就要更多 if, TVT, 面向对象还是没有入门

]]>
发送邮件存在类似于发送短信那样的服务商吗,还是直接用 Python 的 sendmail 函数直接发送就行了 tag:www.v2ex.com,2025-01-23:/t/1107246 2025-01-23T01:57:22Z 2025-01-24T11:14:25Z luoyou1014 https://www.v2ex.com/member/luoyou1014 目前就启用了邮箱的 smtp 功能,然后用 python 的 smtp.sendmail 方法直接发送注册验证码。

但是担心这样大规模发邮件验证码,会被丢进垃圾箱,所以想问一下有没有提供类似于发送短信服务的,可以确保发送邮件验证码不会被扔进垃圾箱的服务。上网搜索了一下,貌似没有看到相关信息。

]]>
Oh-PyTorch: 用 PyTorch 解决 numpy-100 的练习题 tag:www.v2ex.com,2025-01-21:/t/1106932 2025-01-21T17:46:28Z 2025-01-21T20:46:28Z huangyezhufeng https://www.v2ex.com/member/huangyezhufeng TL;DR: https://github.com/shenxiangzhuang/Oh-PyTorch/blob/master/100-PyTorch-Exercises.ipynb

主要是帮助练习和学习 Torch 的一些基本概念和基本操作,希望能对大家有帮助!

]]>
宝塔不能自动续签的 bug 修复 tag:www.v2ex.com,2025-01-21:/t/1106806 2025-01-21T07:32:25Z 2025-01-29T07:25:20Z lyxxxh2 https://www.v2ex.com/member/lyxxxh2 环境

宝塔版本:9.0.0

系统:Ubuntu 24.04.1 LTS (Noble Numbat) x86_64(Py3.7.16)

起因

https://i.imgur.com/Rshnwmp.png

发现证书居然有 25 天的,宝塔自动续签摆设的???

/www/server/panel/pyenv/bin/python3 -u /www/server/panel/class/acme_v2.py --renew=1

acme_v2.py

https://i.imgur.com/OtIPz0a.png

经过测试,发现 p_key 多了个换行。 因此删除换行即可:

if pkey[-1] == '\n':
    pkey = pkey[:-1]

(换行是宝塔加的,因为证书都是从宝塔点击申请的)

结果: https://i.imgur.com/4vwi8LA.png

]]>
关于 py 异步队列 异步支持选择问题 tag:www.v2ex.com,2025-01-19:/t/1106226 2025-01-19T07:27:43Z 2025-01-20T15:30:53Z bwijn https://www.v2ex.com/member/bwijn
老生常谈的话题,异步高并发

存在即合理一个东西存在就是为了解决某种问题。先把问题抛出来: 为什么要用异步?

是为了解决异步 IO 提高性能

关于我的试错和建议:

celery 直接扔了 不要用了

1.由于 decorater 封装 没补全,看着难受

2.monkeypatch 性能问题 和 bug



推荐 arq 快快快 还是快 async 原生支持

对于我这种不想换技术栈的,py 一把梭哈用 arq Redis



如果再追求极致性能用 golang 分开写了 ]]>
py 的异步队列选择 celery arq drm rq ,好像还有一个国人大佬写的(忘了叫啥?名字貌似起的很直白,很牛逼。) tag:www.v2ex.com,2025-01-18:/t/1106168 2025-01-18T17:10:20Z 2025-01-25T20:46:14Z bwijn https://www.v2ex.com/member/bwijn
docker 你真该死啊,太好用了,也太反直觉了,container 没有进程运行竟然启动不起来?非得写 dockerfile ?

为了可扩展 解藕 逼着自己加了 middleware

这几天头都炸了,心里说好的这辈子不用 docker ,后来发现真香

但是 vscode 进 container 里开发竟然要重装插件????

celery 搞得我想砸电脑。

模块名找不到,win 下 fork 又无法运行

arq 文档太少,ai 都没法补全

drm 没用过 rq ???? ]]>
用几行代码轻松构建 gRPC 微服务 tag:www.v2ex.com,2025-01-15:/t/1105399 2025-01-15T16:04:57Z 2025-01-15T17:02:57Z yatao https://www.v2ex.com/member/yatao 在现代分布式系统中,gRPC因其高性能和跨语言支持成为构建微服务的热门选择。然而,传统的 gRPC 开发需要手动编写 .proto 文件和生成代码,这对开发者来说是一项重复且容易出错的任务。FastGRPC 的出现,通过其简洁直观的设计理念,让我们能够用更少的代码和更高的效率构建 gRPC 服务。


1. 安装 FastGRPC

首先,确保你的环境支持 Python 3.9 及以上版本。然后通过以下命令安装 FastGRPC

pip install python-fast-grpc

2. 几行代码实现一个 gRPC 服务

以下是使用 FastGRPC 创建一个简单服务的代码示例:

from pydantic import BaseModel
from fast_grpc import FastGRPC

# 初始化服务
app = FastGRPC(service_name="Greeter", proto="greeter.proto")

# 定义请求和响应的数据模型
class HelloRequest(BaseModel):
    name: str

class HelloReply(BaseModel):
    message: str

# 定义一个 gRPC 方法
@app.unary_unary()
async def say_hello(request: HelloRequest) -> HelloReply:
    return HelloReply(message=f"Greeter SayHello {request.name}")

# 启动服务
if __name__ == '__main__':
    app.run()

代码特点:

  • 使用 Pydantic 直接定义请求和响应的模型,避免了手动解析数据的复杂性。
  • 自动生成 .proto 文件,无需额外工具或命令,也支持根据 proto 自动生成 pydantic 版本的 client 。
  • 通过装饰器注册 gRPC 方法,符合 Python 的直观风格。

通过以上几行代码,我们快速构建了一个基于 gRPC 的 Python 微服务。更多功能,请访问 FastGRPC

]]>
请教一个关于 Python repr 方法的问题 tag:www.v2ex.com,2025-01-14:/t/1105047 2025-01-14T08:41:11Z 2025-01-15T03:14:12Z barantt01 https://www.v2ex.com/member/barantt01 示例代码如下

tmp = '{"{\"username\":\"juheso\",\"is_cache\":1}":3}'
print(tmp)
print(repr(tmp))
tmp_1 = r'{"{\"username\":\"juheso\",\"is_cache\":1}":3}'
print(tmp_1)

我运行之后的结果为

{"{"username":"juheso","is_cache":1}":3}
'{"{"username":"juheso","is_cache":1}":3}'
{"{\"username\":\"juheso\",\"is_cache\":1}":3}

我想要的结果其实是{"{\"username\":\"juheso\",\"is_cache\":1}":3}

但是为什么 repr() 没有效果呢?

我的 python 版本为 3.10

]]>
mac 下使用 pynput 在某些软件(chiaki)下无效 tag:www.v2ex.com,2025-01-11:/t/1104373 2025-01-11T09:56:10Z 2025-01-11T17:41:16Z hyclew https://www.v2ex.com/member/hyclew mac:2018 款 intel python:3.9 os:ventura 13.7.1 chiaki:2.2.0 ,ps 的远程游玩软件,软件本身也是有键盘监控,这个会和 pynput 冲突么?

使用 pynupt 执行键盘模拟,通过 automator 配置服务、系统快捷键绑定脚本。代码如下,正常在编辑器里看执行是正常的,但是打开 chiaki ,快捷键调起脚本后就没反应了,但是如果多调用几次脚本会偶尔还会执行一次按键调用。不知道有没有大佬能指导下?

或者还有其他推荐的自动化工具么?主要需求就是模拟键盘,可以获取图像进行分析比如获取像素点 rgb 或者用 opencv 做图像识别就更好了。

import time from pynput.keyboard import Key, Controller keyboard = Controller() time1=0.1 for num in range(0,120): keyboard.tap("k"); time.sleep(time1); keyboard.tap(Key.left); time.sleep(time1); keyboard.tap("k"); time.sleep(time1); keyboard.tap(Key.down); time.sleep(time1);

]]>
Flask+SQLite+Tornado 在 Windows 2019 下 100 并发都撑不住是怎么回事? tag:www.v2ex.com,2025-01-10:/t/1104276 2025-01-10T14:11:19Z 2025-01-20T00:29:39Z drymonfidelia https://www.v2ex.com/member/drymonfidelia 非常简易的一个测试程序

# -*- coding: UTF-8 -*-

from flask import Flask, request
import json
import time
import re

import sys
import asyncio

from tornado.ioloop import IOLoop
from tornado.wsgi import WSGIContainer
from tornado.httpserver import HTTPServer

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///database.db?check_same_thread=False')

Base = declarative_base()


class ClientLog(Base):
    __tablename__ = 'clientlog'
    log_id = Column(Integer, primary_key=True, autoincrement=True)
    message = Column(String(512))

    def __repr__(self):
        return "A"


from sqlalchemy.ext.declarative import DeclarativeMeta


class AlchemyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            # an SQLAlchemy class
            fields = {}
            for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                data = obj.__getattribute__(field)
                try:
                    json.dumps(data)  # this will fail on non-encodable values, like other classes
                    fields[field] = data
                except TypeError:
                    fields[field] = None
            # a json-encodable dict
            return fields

        return json.JSONEncoder.default(self, obj)


Base.metadata.create_all(engine, checkfirst=True)

Session = sessionmaker(bind=engine)
session = Session()


app = Flask("app")

@app.route('/api', methods=['GET', 'POST'])
@app.route('/api/', methods=['GET', 'POST'])
def api1():
    session.add(ClientLog(message="test"))
    session.commit()
    return "test"

def launch_server():
    if sys.platform == 'win32':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    http_server = HTTPServer(WSGIContainer(app))
    http_server.listen(8080)
    IOLoop.current().start()

launch_server()

在阿里云的 Windows 2019 模板下 100 并发就会崩溃退出,报错

Exception in thread Tornado selector:
Traceback (most recent call last):
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\threading.py", line 982, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\tornado\platform\asyncio.py", line 574, in _run_select
    rs, ws, xs = select.select(to_read, to_write, to_write)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: too many file descriptors in select() 

但是我在 Windows 10 的电脑上测试并发没问题 之前测试换了另外一个对 Windows 更友好的 web framework (想不起来名字,找了半天没找到)可以是可以,但是估计它用了不止一个进程写 SQLite ,导致冲突无法提交更改 业务需求调用一个 Windows Only 的 pip 包 ,无法更换 Linux 服务器

]]>
请教 Python 大神关于 flask 压测 rps 掉到 0 tag:www.v2ex.com,2025-01-10:/t/1104209 2025-01-10T08:53:26Z 2025-01-14T21:29:51Z Flourite https://www.v2ex.com/member/Flourite from flask import Flask app = Flask(__name__) @app.route("/") def hello_world(): return "Hello, World!"

以上代码使用 wrk 压测(wrk -t2 -c100 -d10s http://localhost:5000/ ),第三次 rps 就会掉到 0 ,是什么原因导致的?是 GIL 锁导致的吗?

]]>
想请问在 his 系统中病历号生成规则是什么? tag:www.v2ex.com,2025-01-10:/t/1104205 2025-01-10T08:30:33Z 2025-01-10T13:18:19Z longmeier90 https://www.v2ex.com/member/longmeier90 去医院挂号,感觉 ID 号是有规律的但是就是猜不到什么规律,有人知道吗。或有现成规则生成的公式吗
``` ]]>
请教 Python 大佬 Django-Q 的使用问题 tag:www.v2ex.com,2025-01-08:/t/1103468 2025-01-08T03:28:39Z 2025-01-08T15:28:45Z xuqiccr https://www.v2ex.com/member/xuqiccr ​ 小弟在使用 Django-Q 的时候发现建了 100 个任务,但是 q_cluster 只会执行前几十个任务,后面的任务会在队列中一直到 timeout 结束,但是看进程 q_cluster 的进程又是一直存在,头秃,不知道有大佬可以指点一下吗?

我的 settings:

# djangoQ 配置
Q_CLUSTER = {
    'name': 'myDjangoQ',  # 启动服务名
    'workers': 2,  # 多少个 workers 的数量直接影响了任务处理的并发能力
    'label': 'myDjangoQ_label',
    'orm': 'default',  # 使用 Django ORM 作为后端
    # 'recycle': 4,  # 工作进程在处理完指定数量的任务后,将自动重启自身以释放内存资源
    'timeout': 10,  # 超时
    # 'recycle_frequency': 4,  # 重启频率
    'compress': False,  # 是否将任务包压缩后发送到代理
    'save_limit': 250,  # 保存成功任务结果
    'sync': False,  # 是否同步执行任务
    'queue_limit': 2,  # 集群中的任务数量
    'cpu_affinity': 1,  # 单个任务使用 cpu 核心
    "redis": {
        "host": config.get("redis", 'host'),
        "port": config.get("redis", 'port'),
        "db": 3,
        "password": config.get("redis", 'password'),
        "socket_timeout": 30,
        "charset": "utf-8",
        "decode_responses": True,
        "max_connections": 1000,
    }
}

触发函数和具体执行函数:

def test(request):
    from django_q.tasks import async_task, result
    info = {'code': 0, 'data': [], 'msg': ''}
    try:
        task_ids = []
        for i in range(700):
            task_id = async_task(my_function, i)  
            task_ids.append(task_id)
    except:
        info['msg'] = 'test error'
        logger.error(traceback.format_exc())
    return JsonResponse(info, safe=False)

def my_function(i):
    logger.info('i:{} 开始开始开始开始开始'.format(i))
    logger.info('i:{} 开始'.format(i))
    time.sleep(random.randint(3, 5))
    logger.info('i:{} 结束'.format(i))
    return i

实际输出:

11:15:31 [Q] INFO Process-1:1 processing [undress-hot-october-high]
2025-01-08 11:15:31.602 | INFO     | deploy_queue.views:my_function:1637 - i:20 开始开始开始开始开始
2025-01-08 11:15:31.602 | INFO     | deploy_queue.views:my_function:1643 - i:20 开始
2025-01-08 11:15:31.610 | INFO     | deploy_queue.views:my_function:1645 - i:19 结束
11:15:31 [Q] INFO Process-1:2 processing [network-pizza-sink-emma]
2025-01-08 11:15:31.611 | INFO     | deploy_queue.views:my_function:1637 - i:21 开始开始开始开始开始
2025-01-08 11:15:31.611 | INFO     | deploy_queue.views:my_function:1643 - i:21 开始
11:15:31 [Q] INFO Processed [tennessee-sierra-timing-michigan]
11:15:31 [Q] INFO Processed [arkansas-muppet-charlie-orange]
2025-01-08 11:15:35.615 | INFO     | deploy_queue.views:my_function:1645 - i:21 结束
2025-01-08 11:15:36.607 | INFO     | deploy_queue.views:my_function:1645 - i:20 结束
11:15:41 [Q] WARNING reincarnated worker Process-1:1 after timeout
11:15:41 [Q] INFO Process-1:5 ready for work at 30020
11:15:41 [Q] INFO Process-1:5 processing [zulu-october-green-berlin]
2025-01-08 11:15:41.873 | INFO     | deploy_queue.views:my_function:1637 - i:22 开始开始开始开始开始
11:15:42 [Q] WARNING reincarnated worker Process-1:2 after timeout

可以看到前面 21 个任务执行正常,但是第 22 个开始就全是 timeout 退出,完全没什么头绪,令人头大。

]]>
fastapi 中有没有好用的 orm 工具推荐,以及 orm 是方便迁移的,可以实现像 django 的 migrate 的功能 tag:www.v2ex.com,2025-01-08:/t/1103368 2025-01-08T01:01:38Z 2025-01-08T20:25:26Z lzyong2019 https://www.v2ex.com/member/lzyong2019 有类似推荐的项目也可。

]]>
Windows 新版本怎么实现截图 tag:www.v2ex.com,2025-01-07:/t/1103314 2025-01-07T13:10:56Z 2025-01-07T15:15:30Z xueyuehua https://www.v2ex.com/member/xueyuehua 以前写过一个 Python 脚本,用 pywin32 利用窗口句柄实现内存截图,在 win10 上可以正常使用,窗口被遮挡也可以截图,甚至可以在后台操作鼠标。 但是为什么在 win11 上用不了了,用 c# 实现也是不行,是 win11 新内核导致的吗? 有大佬知道什么解决方案吗

]]>
最近想搞个爬虫爬点东西, 目前 Python 那个框架最好用呢 tag:www.v2ex.com,2025-01-05:/t/1102662 2025-01-05T06:36:15Z 2025-01-14T15:19:10Z iorilu https://www.v2ex.com/member/iorilu 都说 python 最擅长爬虫

但实际上我以前没啥需求, 没怎么实践过

请推荐下实际生产项目中好用的爬虫框架或方案

]]>
Python 打包 exe 方案 tag:www.v2ex.com,2025-01-04:/t/1102497 2025-01-04T07:23:25Z 2025-01-08T09:55:22Z lone6 https://www.v2ex.com/member/lone6 平时会开发一些小脚本,在工作中使用

想请教一下,打包 exe 的方案有哪些?

目前在使用 pyinstaller ,但是打包过程比较慢,打包后的 exe 感觉也挺大的

有没有更好的方案或者优化教程?

]]>