Python基于 BaseHTTPRequestHandler 创建简单Web服务

启动一个最基础的 WEB 服务

创建文件 server.py

# Python 3 server example
from http.server import BaseHTTPRequestHandler, HTTPServerhostName = "localhost"
serverPort = 8080class MyServer(BaseHTTPRequestHandler):def do_GET(self):self.send_response(200)self.send_header("Content-type", "text/html")self.end_headers()self.wfile.write(bytes("<html><body><p>Request: %s</p></body></html>" % self.path, "utf-8"))if __name__ == "__main__":webServer = HTTPServer((hostName, serverPort), MyServer)print("Server started http://%s:%s" % (hostName, serverPort))try:webServer.serve_forever()except KeyboardInterrupt:passwebServer.server_close()print("Server stopped.")

启动命令

python3 server.py

区分访问路径

do_GET方法内, 使用 self.path 变量区分

def do_GET(self):if self.path == '/':self.send_response(200)self.send_header("Content-type", "text/html")self.end_headers()self.wfile.write(bytes("<html><body><p>Request: %s</p></body></html>" % self.path, "utf-8"))elif self.path == '/upload':self.send_response(200)self.send_header("Content-type", "text/html")self.end_headers()self.wfile.write(bytes("<html><body><p>Request: %s</p></body></html>" % self.path, "utf-8"))

处理 POST 请求

实现do_POST方法

def do_POST(self):content_length = int(self.headers['Content-Length'])file_content = self.rfile.read(content_length)# Do what you wish with file_content#print file_content# Respond with 200 OKself.send_response(200)self.send_header("Content-type", "text/html")self.end_headers()self.wfile.write(bytes("<html><body><p>Request: %s</p></body></html>" % self.path, "utf-8"))

处理请求参数和 COOKIE 等

添加以下引用

from functools import cached_property
from http.cookies import SimpleCookie
from urllib.parse import parse_qsl, urlparse

在 MyServer 类下添加以下处理方法

    @cached_propertydef url(self):return urlparse(self.path)@cached_propertydef query_data(self):return dict(parse_qsl(self.url.query))@cached_propertydef post_data(self):content_length = int(self.headers.get("Content-Length", 0))return self.rfile.read(content_length)@cached_propertydef form_data(self):return dict(parse_qsl(self.post_data.decode("utf-8")))@cached_propertydef cookies(self):return SimpleCookie(self.headers.get("Cookie"))

处理 Multipart 文件上传

需要引入

from urllib.parse import parse_qs, parse_qsl, urlparse
import cgi

对请求根据 content-type 分别处理

    def parse_POST(self):print(self.headers)ctype, pdict = cgi.parse_header(self.headers['content-type'])if ctype == 'multipart/form-data':print("file request")pdict['boundary'] = bytes(pdict['boundary'], "utf-8")postvars = cgi.parse_multipart(self.rfile, pdict)elif ctype == 'application/x-www-form-urlencoded' or 'application/json':   print("non-file request")length = int(self.headers['content-length'])postvars = parse_qs(self.rfile.read(length).decode('utf8'),keep_blank_values=1)elif ctype == 'application/octet-stream':print("octet stream header")postvars = {}else:print("nothing")postvars = {}a = self.rfileprint(dir(a))print(a.peek())return postvars

do_POST 中调用

    def do_POST(self):postvars = self.parse_POST()print(postvars)

一个接收文件并调用 PaddleOCR 识别的WEB服务例子

server.py

from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, parse_qsl, urlparse
import cgi
import json
import paddleocr_helperhostName = "localhost"
serverPort = 8080class MyServer(BaseHTTPRequestHandler):def parse_POST(self):print(self.headers)ctype, pdict = cgi.parse_header(self.headers['content-type'])if ctype == 'multipart/form-data':print("file request")pdict['boundary'] = bytes(pdict['boundary'], "utf-8")postvars = cgi.parse_multipart(self.rfile, pdict)elif ctype == 'application/x-www-form-urlencoded' or 'application/json':   print("non-file request")length = int(self.headers['content-length'])postvars = parse_qs(self.rfile.read(length).decode('utf8'),keep_blank_values=1)elif ctype == 'application/octet-stream':print("octet stream header")postvars = {}else:print("nothing")postvars = {}a = self.rfileprint(dir(a))print(a.peek())return postvarsdef do_GET(self):if self.path == '/':self.send_response(200)self.send_header("Content-type", "text/html")self.end_headers()self.wfile.write(bytes("<html><body><p>Request: %s</p></body></html>\r\n" % self.path, "utf-8"))def do_POST(self):postvars = self.parse_POST()#print(postvars)#print(type(postvars['file']))result = {}try:result = paddleocr_helper.parse_and_lookup(postvars['file'][0])except:print("Error occurred")result = {"code": 1,"message": "error","data": None}pass# Respond with 200 OKself.send_response(200)self.send_header("Content-type", "application/json")self.end_headers()#self.wfile.write(bytes("<html><body><p>Request: %s</p></body></html>\r\n" % self.path, "utf-8"))self.wfile.write(bytes(json.dumps(result), "utf-8"))if __name__ == "__main__":webServer = HTTPServer((hostName, serverPort), MyServer)print("Server started http://%s:%s" % (hostName, serverPort))try:webServer.serve_forever()except KeyboardInterrupt:passwebServer.server_close()print("Server stopped.")

paddleocr_helper.py

import json
import re
import math
from paddleocr import PaddleOCRdef parse_image(imagedata):ocr = PaddleOCR(show_log=False, use_angle_cls=True, lang="ch") # need to run only once to download and load model into memoryresult = ocr.ocr(imagedata, cls=True)outputs = []for idx in range(len(result)):res = result[idx]output = []for line in res:ocr_result = {'boxes' : line[0],'text'  : line[1][0],'score' : line[1][1]}output.append(ocr_result)outputs.append(output)return outputsdef lookup_invoice_number(ocr_blocks):block_no = Noneblock_numbers = []for ocr_block in ocr_blocks:#print(ocr_block['text'])regex_result = re.compile('(票据号码|柔据号码|桑据号码|柔线号码|柔楼号码|系热号码):(\d+)').search(ocr_block['text'])if not regex_result is None:#print('----> ' + regex_result.group(2))return regex_result.group(2)regex_result = re.compile('N(?:.*?)(\d{8,})').search(ocr_block['text'])if not regex_result is None:#print('----> ' + regex_result.group(1))return regex_result.group(1)if re.match('No(\.|:|:)', ocr_block['text']):#print('- No. block: {}'.format(ocr_block['text']))block_no = ocr_blockregex_result = re.compile('(\d{8,})').search(ocr_block['text'])if not regex_result is None:#print('- Num block: {}'.format(regex_result.group(1)))ocr_block['text'] = regex_result.group(1)block_numbers.append(ocr_block)if not block_no is None and not len(block_numbers) == 0:#print('- block_no:{}'.format(block_no))distance_min = Nonecandidate = Nonefor block_number in block_numbers:# calculate distance between number and Nodistance = calcu_distance(block_no['boxes'], block_number['boxes'])#print('- dist:{}, block:{}'.format(distance, block_number))print('- dist:{}, block:{}'.format(distance, block_number['text']))if (distance_min is None) or (distance < distance_min):distance_min = distancecandidate = block_number['text']return candidatereturn Nonedef parse_and_lookup(imagedata):invoice_numbers = []ocr_result = parse_image(imagedata)if len(ocr_result) > 0:for ocr_blocks in ocr_result:invoice_number = lookup_invoice_number(ocr_blocks)if (not invoice_number is None):invoice_numbers.append(invoice_number)resp = {"code": 0,"message": "succ","data": {"invoice_numbers": invoice_numbers}}return respdef calcu_distance(boxes1, boxes2):distance_min = Nonebox1 = Nonefor box in boxes1:distance = point_distance(box, boxes2[0])if (distance_min is None) or (distance < distance_min):distance_min = distancebox1 = boxfor box in boxes2:distance = point_distance(box, box1)if (distance_min is None) or (distance < distance_min):distance_min = distancereturn distance_mindef point_distance(point1, point2):x = point1[0] - point2[0]y = point1[1] - point2[1]qrt = math.sqrt(x**2 + y**2)return qrt

参考

  • https://realpython.com/python-http-server/

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/773072.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

哈佛毕业生考公,大材小用?

阅读本文大概需要 1.25 分钟 前段时间一张杭州市余杭区公务员录用名单的截图在网络上疯传&#xff0c;进入公示名单的人中&#xff0c;不乏许多毕业于名校的学生&#xff0c;甚至有些是哈佛毕业的。。。许多人对此感到惊讶&#xff0c;认为连哈佛毕业生都参加公务员考试&#x…

Linux文件和文件夹操作

一、文件操作 功能项命令实例作用文件创建vi /opt/learn/hello.txt 在目录/opt/learn下创建文件hello.txt并进入vi编辑界面 touch /opt/learn/test在目录/opt/learn下创建空白文件testcat > /opt/catfile创建文件catfile并在屏幕上输入内容&#xff0c;最后按 Ctrl D 退出…

压力测试(QPS)及测试工具Locust

压力测试&#xff1a; 通常指的是确定接口或服务能够处理的最大请求量&#xff08;吞吐量&#xff09;和并发用户数&#xff0c;同时保持合理的响应时间和稳定性。 性能目标 最大吞吐量&#xff1a;系统每秒可以处理的请求数。最大并发用户数&#xff1a;系统可以同时支持的…

【学习心得】神经网络知识中的符号解释

这里我对我学到的神经网络知识中&#xff0c;常见的符号做一下记录和总结&#xff0c;方便自己在后面学习中复习。下图二分类识别图像识别猫为例。为了保存一张图片&#xff0c;需要三个矩阵&#xff0c;它们分别对应图片中的红、绿、蓝三种颜色通道&#xff0c;如果图片大小为…

【0278】checkpointer 共享内存(CheckpointerShmem)初始化(3)

0. 关于checkpointer 检查指针是Postgres 9.2的新特性。它处理所有检查点。自上次检查点以来,检查点在经过一定时间后自动分发,并且还可以发出信号来执行请求的检查点。(GUC参数要求每隔这么多WAL段就有一个检查点,这是通过后端在填充WAL段时发出信号来实现的; checkpointer…

ubuntu22.04安装k3s, kuboard

直接安装方式 更新系统包列表 sudo apt update安装所需的依赖 sudo apt install -y curl下载k3s安装脚本 curl -sfL https://get.k3s.io | sh -验证k3s安装 sudo systemctl status k3s获取k3s的token sudo cat /var/lib/rancher/k3s/server/node-token设置kubectl配置 mkdir -p…

Django路由

Router介绍 在实际开发过程中&#xff0c;一个Django项目会包含很多的app,这时候如果我们只在主路由里进行配置就会显得杂乱无章&#xff0c;所以通常会在每个app里&#xff0c;创建各自的urls.py路由模块&#xff0c;然后从根路由出发&#xff0c;将app所属的url请求&#xff…

类与对象上C++

文章目录 一、1.面向过程和面向对象初步认识二、类的定义1.类的两种定义方式 三、类的访问限定符及封装1.访问限定符2.面试题3.封装 四、类对象模型1. 类对象的存储方式猜测2. 结构体内存对齐规则 五、this指针1.this指针的引出2.this指针的特性 六、 C语言和C实现Stack的对比 …

【蓝桥杯3.23小白赛】(详解)

第一题签到题不多说 【二进制王国】 #include <iostream> #include <vector> #include <algorithm> using namespace std;//int Cmp(string s1, string s2)测试了一下时间差确实很明显&#xff0c;还是用下面的内个 int Cmp(const string &s1,const st…

如何快速进行城市内涝模拟?HTWATER软件

原文链接&#xff1a;如何快速进行城市内涝模拟&#xff1f;HTWATER软件https://mp.weixin.qq.com/s?__bizMzUzNTczMDMxMg&mid2247599079&idx2&sndc6f3da8b17c5587cf5b7766e7019729&chksmfa820200cdf58b16658983ecfbf2b369bff39813302942d6f7eb7b71428c68da71…

在 ubuntu server 22 上安装 Docker 引擎

在 ubuntu server 22 上安装 Docker 引擎 在开始之前&#xff0c;请确保你的系统满足以下条件&#xff1a; 64 位版本的 Ubuntu Mantic 23.10、Ubuntu Jammy 22.04 (LTS) 或 Ubuntu Focal 20.04 (LTS)使用 sudo 或 root 权限 安装步骤 1. 设置 Docker 的 apt 仓库 首先&am…

在Windows上交叉编译STM32(环境搭建)

在Windows上交叉编译STM32 Keil 虽然好用&#xff0c;但是是收费的&#xff0c;不想破解怎么办~ 使用交叉编译工具&#xff01; 交叉编译工具下载 官方交叉编译工具下载连接 下载解压好后将 bin 目录写入 PATH&#xff0c; 使用命令行检测是否安装成功。 Windows 安装 make …

未解决的问题:使用fgets循环读取字符串总是少一行

一、使用fgets循环读取字符串总是少一行 代码 #include<stdio.h> int main() {int n, i;scanf("%d", &n);char line[n][1000];for (i 0; i < n; i) {fgets(line[i], 1000, stdin);}for (i 0; i < n; i) {printf("%s", line[i]);}retur…

纳斯达克大屏媒体尺寸与投放费用:一次投放需要多少钱?

纳斯达克大屏媒体尺寸与投放费用&#xff1a;一次投放需要多少钱&#xff1f; 1. 纳斯达克图片要求 1.1 像素要求 高度&#xff1a;2336 像素宽度&#xff1a;1832 像素 1.2 分辨率要求 像素比率&#xff1a;1.0 px 72 dpi 1.3 文件格式要求 静态图片格式&#xff1a;.…

Django 仿博客园练习

数据库搭建 部分功能介绍 【一】注册 &#xff08;1&#xff09;效果显示、简单简介 主要亮点 结合了layui和forms组件默认头像可以随着性别的选择发生改变自定义头像可以实时更新显示forms组件报错信息可以局部刷新显示在对应框体下面 没有直接使用layui的前端验证后端验证…

unity中手势识别开源代码——HandPoseBarracuda

HandPoseBarracuda是一个使用单目彩色摄像头工作的神经网络手/手指追踪器的概念验证实现。 基本上,HandPoseBarracuda是MediaPipe Hands管道的一个部分端口。尽管它并不是原始包的直接端口,但它使用了相同的基本设计和相同的预训练模型。 请注意,这只是一个概念验证实现。…

SQLite数据库成为内存中数据库(三)

返回&#xff1a;SQLite—系列文章目录 上一篇&#xff1a;SQLite使用的临时文件&#xff08;二&#xff09; 下一篇&#xff1a;SQLite中的原子提交&#xff08;四) ​​ SQLite数据库通常存储在单个普通磁盘中文件。但是&#xff0c;在某些情况下&#xff0c;数据库可能…

Dockerfile和Docker-compose

一、概述 Dockerfile和Docker Compose是用于构建和管理 Docker 容器的两个工具&#xff0c;但它们的作用和使用方式不同。 Dockerfile Dockerfile 是一个文本文件&#xff0c;用于定义 Docker 镜像的构建规则。它包含一系列指令&#xff0c;如 FROM&#xff08;指定基础镜像…

excel统计分析——协方差分析基本原理

参考资料&#xff1a;生物统计学 方差分析中&#xff0c;要求除试验因素外的其他条件保持在相同水平上才能对实验结果的差异显著性进行比较&#xff0c;然而有些非试验因素很难或不可能人为控制&#xff0c;此时如果使用方差分析法推断处理其差异显著性&#xff0c;往往会导致错…

python数据解析xpath

前言一、安装&#xff1f;二、使用步骤1.基本使用**【2】谓语&#xff08;Predicates&#xff09;**案例 前言 xpath在Python的爬虫学习中&#xff0c;起着举足轻重的地位&#xff0c;对比正则表达式 re两者可以完成同样的工作&#xff0c;实现的功能也差不多&#xff0c;但xp…