[python] subprocess

proc.py

import asyncio

from argparse import ArgumentParser

from asyncio import sleep

async def process_function(user, file_path):

    if user == 'user1':

        await sleep(2)

    result = f"Process for user {user}"

    with open(file_path, 'w') as fh:

        fh.write(result)

    print("Hello world")

if __name__ == '__main__':

    parser = ArgumentParser()

    parser.add_argument('--name')

    parser.add_argument('--path')

    args = parser.parse_args()

    user = args.name

    file_path = args.path

    asyncio.run(process_function(user, file_path)

=====================================================

main.py

import asyncio

import subprocess

async def proc(user):

    file_path = f"{user}_output.txt"

    process = subprocess.Popen(['python', 'proc.py', '--name', user, '--path', file_path])

    await asyncio.to_thread(process.wait)

    with open(f"{user}_output.txt", "r") as output_file:

        result = output_file.read()

        print(result)

async def broker():

    users = ['user1', 'user2', 'user3']

    await asyncio.gather(*(proc(user) for user in users))

def main():

    asyncio.run(broker())

if __name__ == "__main__":

    main()

[python] tornado 오래 걸리는 작업

 import json

import tornado.ioloop

import tornado.web

from src.exec import Executor

class ShellHandler(tornado.web.RequestHandler):

    async def long_running_task(self):

        """

        get, post 로 호출 시 오래 걸리는 작업.

        """

        executor = Executor()

        stdout, stderr = await executor.run()

        return stdout, stderr

    async def post(self):

        data = self.request.body.decode('utf-8')

        try:

            json_data = json.loads(data)

            self.write(f'received => {json_data}')

        except json.JSONDecodeError:

            self.write(f'error => {data}')

    async def get(self):

        ip_address = self.request.remote_ip

        stdout, stderr = await self.long_running_task()

        self.write(f"{ip_address}<br>stdout={stdout}<br>stdout={stderr}")

def make_app():

    return tornado.web.Application([

        (r"/shell", ShellHandler),

    ])

if __name__ == "__main__":

    app = make_app()

    app.listen(8888)

    print("Server is listening on port 8888")

    tornado.ioloop.IOLoop.current().start()

[python] pdf 를 이미지로 저장

• poppler 다운로드 

https://github.com/oschwartz10612/poppler-windows/releases/tag/v23.11.0-0

• 압축 해제

C:\poppler-23.11.0

• pdf2image 설치

pip install pdf2image

def save_pdf_page_as_image(pdf_path, page_number, filename, output_dir):

    # PDF 파일에서 이미지로 변환
    name = os.path.splitext(filename)[0]
    output_path = os.path.join(output_dir, '{0}.png'.format(name))
    images = convert_from_path(pdf_path, first_page=page_number, last_page=page_number, poppler_path=r'C:\poppler-23.11.0\Library\bin')

    if images:
        # 첫 번째 이미지 선택 (페이지 번호에 해당하는 이미지)
        image = images[0]
        # 이미지 저장
        image.save(output_path, 'PNG')
    else:
        print(f'error.')

tensorflow 호출 시 메모리 사용량이 계속 늘어날 때

tf.keras.backend.clear_session()

python pdf 에 워터마크 추가

출처] https://herohjk.com/41

import os

from PIL import Image

from reportlab.pdfgen.canvas import Canvas

from PyPDF2 import PdfReader, PdfWriter

class Watermark:

    def __init__(self, output_path, image_path, src_path, working_dir):

        self.output_path = output_path

        self.image_path = image_path

        self.src_path = src_path

        self.working_dir = working_dir

        if not os.path.exists(self.working_dir):

            os.makedirs(self.working_dir)

    @staticmethod

    def clear_white_background(image):

        """

        :param image:

        :return:

        """

        image = image.convert('RGBA')

        image_data = image.getdata()

        new_image_data = []

        for pixel in image_data:

            if pixel[0] > 240 and pixel[1] > 240 and pixel[2] > 240:

                new_image_data.append((0, 0, 0, 0))

            else:

                new_image_data.append(pixel)

        image.putdata(new_image_data)

        return image

    @staticmethod

    def image_to_pdf(image_path, pdf_path):

        """

        :param image_path:

        :param pdf_path:

        :return:

        """

        size = Image.open(image_path, 'r').size

        new_canvas = Canvas(pdf_path, pagesize=Image.open(image_path, 'r').size)

        new_canvas.drawImage(image=image_path, x=0, y=0, mask='auto')

        new_canvas.save()

    @staticmethod

    def pdf_merge(save_path, pdf_path, watermark_pdf_path):

        """

        :param save_path:

        :param pdf_path:

        :param watermark_pdf_path:

        :return:

        """

        pdf_file = open(pdf_path, 'rb')

        pdf_reader = PdfReader(pdf_file, strict=False)

        watermark_pdf_file = open(watermark_pdf_path, 'rb')

        watermark_pdf = PdfReader(watermark_pdf_file, strict=False).pages[0]

        pdf_writer = PdfWriter()

        for pageNum in range(len(pdf_reader.pages)):

            page_obj = pdf_reader.pages[pageNum]

            x = (page_obj.mediabox[2] - watermark_pdf.mediabox[2]) / 2

            y = (page_obj.mediabox[3] - watermark_pdf.mediabox[3]) / 2

            watermark_pdf.add_transformation(Transformation().translate(tx=x, ty=y))

            page_obj.merge_page(page2=watermark_pdf, expand=False)

            pdf_writer.add_page(page_obj)

        result_file = open(save_path, 'wb')

        pdf_writer.write(result_file)

    def convert(self):

        """

        :return:

        """

        image = Image.open(self.image_path, 'r')

        clear_image = self.clear_white_background(image)

        clear_image_path = os.path.join(self.working_dir, 'clear_image.png')

        clear_image.save(clear_image_path)

        watermark_pdf_path = os.path.join(self.working_dir, 'watermark_pdf.pdf')

        self.image_to_pdf(clear_image_path, watermark_pdf_path)

        self.pdf_merge(self.output_path, self.src_path, watermark_pdf_path)

python tornado 업로드 파일 사이즈.

용량이 큰 파일 업로드 시

net::ERR_CONNECTION_RESET 오류 발생할 경우.

http_server = tornado.httpserver.HTTPServer(app, max_buffer_size=10485760000)

upload_form.html

<!DOCTYPE html>
<html>
<head>
    <title>File Upload Form</title>
</head>
<body>
    <h2>File Upload Form</h2>
    <form action="/" method="post" enctype="multipart/form-data">
        <label for="file">Select a file:</label>
        <input type="file" name="file" id="file"><br>
        <input type="submit" value="Upload">
    </form>
</body>
</html>

main.py

import os
import tornado.web
import tornado.ioloop

UPLOAD_DIR = "uploads"  # Define the directory where uploaded files will be saved


class FileUploadHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("upload_form.html")  # Render the HTML form for file uploads

    def post(self):
        file_info = self.request.files['file'][0]  # Get file information from the request

        # Extract relevant file information
        file_name = file_info['filename']
        file_body = file_info['body']

        # Save the uploaded file to the specified directory
        file_path = os.path.join(UPLOAD_DIR, file_name)
        with open(file_path, 'wb') as file:
            file.write(file_body)

        self.write(f"File '{file_name}' uploaded successfully.")


def make_app():
    return tornado.web.Application([
        (r"/", FileUploadHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    print("Server is running on http://localhost:8888")
    http_server = tornado.httpserver.HTTPServer(app, max_buffer_size=10485760000)
    http_server.listen(8888)
    tornado.ioloop.IOLoop.current().start()

Python을 Windows service로 실행하기

pywin32 설치

python -m pip install --upgrade pywin32 

- 가상환경에서 설치하면 안됨.

가상 환경 외부에서는 COM 객체, 서비스 등을 설치하고자 할 때

python Script/pywin32_postinstall.py -install

[myservice.py]

import win32serviceutil
import ctypes
import time

OutputDebugString = ctypes.windll.kernel32.OutputDebugStringW


class MyService(win32serviceutil.ServiceFramework):
_svc_name_ = 'MyService'
_svc_display_name_ = 'My Service'
is_running = False

def SvcStop(self):
OutputDebugString("MyService __SvcStop__")
self.is_running = False

def SvcDoRun(self):
self.is_running = True
while self.is_running:
OutputDebugString("MyService __loop__")
time.sleep(1)


if '__main__' == __name__:
win32serviceutil.HandleCommandLine(MyService)

관리자 권한으로 PowerSell 실행

python myservice.py install

python myservice.py start

python myservice.py stop