[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 으로 엑셀 셀 배경색 넣기.

color = 'ffffff'

bc = PatternFill(fgColor=Color(color, tint=0.5), bgColor=Color(color, tint=0.5), fill_type='gray125')

docker 로 django 서비스

•  Dockerfile 파일 생성

FROM python:3.8.10

RUN apt -y update

WORKDIR /mysite

COPY . .

RUN pip install django==4.0.4

CMD ["python3", "manage.py", "runserver", "0:8000"]

EXPOSE 8000

 

• docker build -t django-test .
• docker run -p 8000:8000 -tid django-test

Python 에서 .ui 파일 불러오기 (PyQt5)

import sys

from PyQt5 import uic

from PyQt5.QtWidgets import QMainWindow, QApplication

form_class = uic.loadUiType('./window_main.ui')[0]

class MyWindow(QMainWindow, form_class):

    def __init__(self):

        super().__init__()

        self.setupUi(self)

if __name__ == "__main__":

    app = QApplication(sys.argv)

    myApp = MyWindow()

    myApp.show()

    app.exec_()

excel에 첨부된 이미지 추출.

# -*-coding:utf-8-*-

from openpyxl import load_workbook
from openpyxl_image_loader import SheetImageLoader


def main():
    filepath = './excel/sample.xlsx'
    load_wb = load_workbook(filepath, data_only=True)
    for sheet_name in load_wb.sheetnames:
        sheet = load_wb[sheet_name]

        image_loader = SheetImageLoader(sheet)
        # image = image_loader.get('C3')
        # image.show()
        print(sheet.max_row, sheet.max_column)
        for column in range(sheet.max_column):
            column = chr(ord('A') + column)
            for row in range(sheet.max_row):
                cell_id = '{0}{1}'.format(column, row)
                print(cell_id)
                if image_loader.image_in(cell_id):
                    print(cell_id, '->', image_loader.image_in(cell_id))


if __name__ == '__main__':
    main()

[Python] python 으로 프로그램 목록 구하기

import win32com.client 
strComputer = "." 
objWMIService = win32com.client.Dispatch("WbemScripting.SWbemLocator") 
objSWbemServices = objWMIService.ConnectServer(strComputer,"root\cimv2") 
colItems = objSWbemServices.ExecQuery("Select * from Win32_Product") 
for objItem in colItems: 
    print "Caption: ", objItem.Caption 
    print "Description: ", objItem.Description 
    print "Identifying Number: ", objItem.IdentifyingNumber 
    print "Install Date: ", objItem.InstallDate 
    print "Install Date 2: ", objItem.InstallDate2 
    print "Install Location: ", objItem.InstallLocation 
    print "Install State: ", objItem.InstallState 
    print "Name: ", objItem.Name 
    print "Package Cache: ", objItem.PackageCache 
    print "SKU Number: ", objItem.SKUNumber 
    print "Vendor: ", objItem.Vendor 
    print "Version: ", objItem.Version 

[python] pyqt5 isSignalConnected

특정 시그널에 연결 되어 있는지를 확인

def isSignalConnected(_obj, signal_name):   

    metaObj = _obj.metaObject()

    for i in range(metaObj.methodCount()):      

        meta_method = metaObj.method(i)

        if meta_method.methodType() == QMetaMethod.Signal:         

            if meta_method.name().data().decode('utf8') == signal_name:            

                return True   

    return False

[python] eval

def hello():

    print('Hello, World!')

v = 'hello()'

eval(v)

v = 'hello'

eval(v)()

[Python] Python 에서 java 호출

pip install JPype1

#!/usr/bin/env python# -*- coding:utf-8 -*-
import jpype

if __name__ == '__main__':
    classpath = './jar/whoareyou.jar'    jpype.startJVM(jpype.getDefaultJVMPath(), '-ea', "-Djava.class.path=%s" % classpath)
    WhoAreYou = jpype.JClass("kr.co.demo.WhoAreYou")

    obj = WhoAreYou()
    ret = obj.whatyourname('hello')
    print ret

    WhoAreYou2 = jpype.JPackage("kr").co.demo.WhoAreYou
    obj2 = WhoAreYou()
    ret = obj.whatyourname('world')
    print ret
    jpype.shutdownJVM()

python logging

import logging

logger = logging.getLogger('tornado.general')
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter([%(levelname)s | %(asctime)s | %(filename):%(lineno)04s | %(message)s")
fileHandler = logging.handlers.RotatingFileHandler(filename='', maxBytes=1, backupCount=1)
fileHandler.setFormatter(formatter)
fileHandler.set_name('myname')
logger.addHandler(fileHandler)

logger = logging.getLogger('tornado.general')
hs = logger.handlers[:]
for x in hs:
    if x.name == 'myname':
        logger.handlers.remove(x)

python decode mime

import email
import base64

def decode_mime(_msg, _result=[]):
    def _decode(msg, result):
        if msg.is_multipart():
            for item in msg.get_payload():
                _decode(item, result)
        else:
            if 'content-type' in msg:
                _content_type = msg['content-type']
                if 'text/html' in _content_type:
                    decoded_text = base64.decodestring(msg.get_payload())
                    result.append(decoded_text)
    _decode(_msg, _result)

decode_mime(email.message_from_string(_message), result=[])

python HTMLParser

from HTMLParser import HTMLParser

class MyHTMLParser(HTMLParser):
    def __init__(self):
        HTMLParser.__init(self)
        self.result = list()

    def handle_date(self, data):
        if data is not None:
            self.result.append(data)

obj = MyHTMLParser()
obj.feed(html)
print ' '.join(obj.result)

Tornado 에서 static 파일 (javascript, css) 경로 설정

========= 디렉토리 구조 ====================================
Tornado
         |- static
                  |- script (javascript)
                  |- styles (css)
         |- templates
                  |- index.html

========= python 소스   ====================================

import os
import tornado.ioloop
import tornado.web
 
class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("index.html")
 
def make_app():
    return tornado.web.Application([
        (r"/", MainHandler),
        (r"/(.*)", tornado.web.StaticFileHandler, {'path': os.path.join(os.path.dirname(__file__), 'static')}),
        ],
        template_path=os.path.join(os.path.dirname(__file__), "templates"))
 
if __name__ == '__main__':
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

========= index.html  ====================================

<head>

         <script src="scripts/test.js"></script>

         <link rel="stylesheet" href="styles/main.css" />

</head>

[Python] cx_Oracle CLOB, BLOB 처리

for _row in cur:

    if isinstance(elem, cx_Oracle.LOB) or isinstance(elem, cx_Oracle.CLOB) or isinstance(
elem, cx_Oracle.BLOB):
data = elem.read()

[python] datetime to timestamp, timestamp to datetime convert

import time
from datetime import datetime

if __name__ == '__main__':
   _now = datetime.now()
   print _now
   _timestamp = time.mktime(_now.timetuple())
   print datetime.fromtimestamp(_timestamp)