Интеграция FAST API и ZennoPoster: запуск и вызов функций из Python

volody00

Client
Регистрация
06.09.2016
Сообщения
982
Благодарностей
1 082
Баллы
93

133851

В прошлом конкурсе @KolkaPetkinSyn показал нам, как можно использовать python в зеннопостер с помощью exe файлов. В этой же статье я хочу показать еще один способ с разворачиванием своего API. Это гораздо удобнее (по моему мнению) если вам надо не просто вызывать какую-то программу, а дергать конкретные функции.

Суть в следующем:
  • Пишем какие-то методы на python с помощью нейросетей
  • Разворачиваем API на python
  • В результате будем посылать запросы вроде http://127.0.0.1:8000/upper/?name={-Variable.name-}, а на выходе получать результат в переменную

Пишем код в deepseek

Я решил продемонстрировать все на простом примере. У нас будет программа, где мы на входе подаем pdf файл, а на выходе получаем запароленный pdf.

Иду в deepseek и пишу промпт: “с помощью библиотеки PyPDF2 напиши функцию, которая будет добавлять пароль на pdf файл”

Открываю vs code, открываю терминал (вверху Terminal -> New Terminal) и устанавливаю библиотеку
pip install PyPDF2



Создаем файл test.py (правая кнопка мыши по левой панели -> new file) и вставляем код:

Python:
from PyPDF2 import PdfReader, PdfWriter

def add_password_to_pdf(input_pdf_path: str, output_pdf_path: str, password: str) -> None:
    """Ваша функция для добавления пароля к PDF"""
    reader = PdfReader(input_pdf_path)
    writer = PdfWriter()

    for page in reader.pages:
        writer.add_page(page)

    writer.encrypt(password)

    with open(output_pdf_path, "wb") as f:
        writer.write(f)

add_password_to_pdf('Python.pdf', 'output.pdf', '123')
Запускаем его прописав в терминале
python [URL='http://test.py?roistat_visit=1569852']test.py[/URL]
Как убедились, что все работает, идем поднимать API.

Реализуем API


Использовать будем Fast API.
Устанавливаем библиотеки
pip install fastapi uvicorn PyPDF2 python-multipart
Давайте сначала разберемся на простом примере. Создаем файл testing.py и пишем свой первый код:
Python:
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse

app = FastAPI()

# вот эту часть меняем, остальное просто копируем
@app.get("/upper/", response_class=PlainTextResponse)
async def upper_text(name: str):
    return name.upper()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
Запускаем в терминале. Если у вас как на скрине ниже то сервер успешно запустился

Теперь идем в зеннопостер, создаем кубик GET

Вот и все! Теперь мы можем посылать GET запрос, передавая какое-то значение, а на выходе получаю строку в верхнем регистре.

А как нам использовать написанную ранее add_password_to_pdf? На самом деле все, что нам нужно - это вместо return name.upper() вызвать add_password_to_pdf().

Код:
Python:
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from PyPDF2 import PdfReader, PdfWriter

def add_password_to_pdf(input_pdf_path: str, output_pdf_path: str, password: str) -> None:
    """Ваша функция для добавления пароля к PDF"""
    reader = PdfReader(input_pdf_path)
    writer = PdfWriter()

    for page in reader.pages:
        writer.add_page(page)

    writer.encrypt(password)

    with open(output_pdf_path, "wb") as f:
        writer.write(f)

app = FastAPI()

# вот эту часть меняем, остальное просто копируем
@app.get("/upper/", response_class=PlainTextResponse)
async def upper_text(input: str, output: str, password: str):
    add_password_to_pdf(input, output, password)
    return "ok"

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
Кубик:



Перезапускаем сервер (сначала ctrl+C, чтобы остановить, а затем как обычно python testing.py) , пробуем и все работает.

Это если у вас и python и ZennoPoster работают на одном ПК. Если они на разных ПК, то вам нужно сначала из зенки отправить файл на сервер -> там его принять -> создать файл с паролем -> отправить обратно в зеннопостер -> сохранить файл на пк через зеннопостер. Тут мы будем уже использовать POST запрос. Расписывать долго, образец есть в приложенном шаблоне.

На этом все. Надеюсь, статья была полезна. Если что непонятно заснял еще видео, где я еще раз покажу то, что расписал тут уже на примере библиотеки.

 

Вложения

Последнее редактирование модератором:

Asmus003

Client
Регистрация
25.03.2018
Сообщения
297
Благодарностей
68
Баллы
28
а если нужно сделать несколько совершенно разных скриптов, как это реализовать? тут как я понимаю, запуск 1 скрипта с разными параметрами. и тогда что, на скриптов надо поднимать 10 серверов?
 

volody00

Client
Регистрация
06.09.2016
Сообщения
982
Благодарностей
1 082
Баллы
93
а если нужно сделать несколько совершенно разных скриптов, как это реализовать? тут как я понимаю, запуск 1 скрипта с разными параметрами. и тогда что, на скриптов надо поднимать 10 серверов?
Не надо тебе 10 серверов. Я не думаю, что у тебя там скрипты на десятки тысяч строк кода. По файлам и функциям раскидываешь и всё. Deepseek поспрашивай. Или если есть конкретный пример, то можешь сюда скинуть свои скрипты, покажу как
 

Viking01

Client
Регистрация
19.08.2017
Сообщения
240
Благодарностей
179
Баллы
43
В зенке можно подключать проекты visual studio, а в проекте на c# есть уже готовые методы по выполнению питоновского кода.
и этот проект visual studio потом компилируется в одну dll для использования в зенке.
Не сочти за критику ни разу, это еще один вариант реализации твоей задумки)
 
  • Спасибо
Реакции: Metrix, bvbfor и volody00

Asmus003

Client
Регистрация
25.03.2018
Сообщения
297
Благодарностей
68
Баллы
28
Не надо тебе 10 серверов. Я не думаю, что у тебя там скрипты на десятки тысяч строк кода. По файлам и функциям раскидываешь и всё. Deepseek поспрашивай. Или если есть конкретный пример, то можешь сюда скинуть свои скрипты, покажу как
не, скрипты маленькие, просто разные задачи с разными сайтами и данными. я просто не могу понять, как в логике передается что нужно запускать C:\scripts\script1.py arg1 arg2 или же C:\scripts\script2.py arg1 arg2, как это в гет запросе реализовать?
сейчас у меня запускаются скрипты через командную строку с параметрами, но иногда процессы не закрываются. поэтому интересен твой вариант реализации.
 

volody00

Client
Регистрация
06.09.2016
Сообщения
982
Благодарностей
1 082
Баллы
93
не, скрипты маленькие, просто разные задачи с разными сайтами и данными. я просто не могу понять, как в логике передается что нужно запускать C:\scripts\script1.py arg1 arg2 или же C:\scripts\script2.py arg1 arg2, как это в гет запросе реализовать?
сейчас у меня запускаются скрипты через командную строку с параметрами, но иногда процессы не закрываются. поэтому интересен твой вариант реализации.
там нужно модифицировать скрипты, а для этого нужно соображать, что именно они делают. поэтому я и попросил прикрепить их сюда. попробую объяснить, хотя твой случай может быть иным.

сейчас я создал 2 файла .py (2 скрипта), которые работают через "запуск программы".

так, первый файл "cerenada.py"
Python:
import requests
import search_element
from lxml import html, etree
import csv
import sys

if len(sys.argv) != 2: raise Exception('Не верно переданы аргументы')
URL = sys.argv[1]
print(URL)
if 'cerenada.ru' not in URL: raise Exception('wrong url')

response = requests.get(URL)
response.raise_for_status()

XPATH_CARDS = "//ul[contains(@class, 'products')]/li"

XPATH_TITLE = ".//h2/text()"
XPATH_LINK = ".//a[contains(@href, 'product')]/@href"
XPATH_IMAGE = ".//img/@src"  # коллекция

source = html.fromstring(response.text)


cards = search_element.find_elements(source, XPATH_CARDS)

result = []
for card in cards:
    mass = []
    title_card = search_element.find_element_in_element(card, XPATH_TITLE)
    link_card = search_element.find_element_in_element(card, XPATH_LINK)

    images_card_el = search_element.find_elements_in_element(card, XPATH_IMAGE)
    images_card = ':'.join(images_card_el)
    mass.append(title_card)
    mass.append(link_card)
    mass.append(images_card)
    result.append(mass)

with open('result.csv', 'w', encoding='utf-8-sig', newline='') as file:
    writer = csv.writer(file, delimiter=';')
    writer.writerows(result)

print("okey")
2-й файл upper.py
Python:
import sys

if len(sys.argv) != 2: raise Exception('Не верно переданы аргументы')
phrase = sys.argv[1]

result = phrase.upper()
sys.stdout.write(result)
sys.stdout.flush()
доп файл search_element.py для работы cerenada.py
Python:
from lxml import etree, html
from lxml.html import HtmlElement
import requests

def find_elements(source_code: HtmlElement, xpath: str, raiseException: bool = True):
    """
    Поиск коллекции элементов
    """
    elements = source_code.xpath(xpath)
    if len(elements) == 0:
        if raiseException: raise Exception(f'Не найдены элементы: {xpath}')
        else: return None
    return elements

def find_element(source_code: HtmlElement, xpath: str, raiseException: bool = True):
    """
    Поиск одного элемента
    """
    element = find_elements(source_code, xpath)
    if not element:
        if raiseException: raise Exception(f'Не найден элемент: {xpath}')
        else: return None

    if isinstance(element[0], str): return element[0].strip()
    return element[0]

def find_elements_in_element(element, xpath: str, raiseException: bool = True):
    """
    Поиск коллекции элементов внутри другого элемента
    """
    elements = element.xpath(xpath)
    if not elements:
        if raiseException: raise Exception(f'Не найдены элементы внутри другого элмента: {xpath}')
        else: return None
    return elements

def find_element_in_element(element, xpath: str, raiseException: bool = True):
    """
    Поиск одного элемента внутри другого элемента
    """
    el = element.xpath(xpath)
    if not el:
        if raiseException: raise Exception(f'Не найден элемент внутри другого элмента: {xpath}')
        else: return None
    if isinstance(el[0], str): return el[0].strip()
    return el[0]
пример как он запускается на текущий момент (аналог кубика запуск программы):
C#:
System.Diagnostics.Process cmd = new System.Diagnostics.Process();
cmd.StartInfo.FileName = "cmd.exe";
cmd.StartInfo.RedirectStandardInput = true;
cmd.StartInfo.RedirectStandardOutput = true;
cmd.StartInfo.RedirectStandardError = true; // Добавляем перехват stderr
cmd.StartInfo.CreateNoWindow = true;
cmd.StartInfo.StandardOutputEncoding = Encoding.GetEncoding(866);
cmd.StartInfo.StandardErrorEncoding = Encoding.GetEncoding(866); // Кодировка для stderr
cmd.StartInfo.UseShellExecute = false;
cmd.Start();

cmd.StandardInput.WriteLine(String.Format("cd " + project.Variables["py_path"].Value));
cmd.StandardInput.WriteLine(String.Format("python cerenada.py " + project.Variables["arg"].Value));
cmd.StandardInput.Flush();
cmd.StandardInput.Close();

// Читаем вывод и ошибки
string output = cmd.StandardOutput.ReadToEnd();
string error = cmd.StandardError.ReadToEnd();

cmd.WaitForExit();

// Если есть ошибки - выбрасываем исключение
if (!string.IsNullOrEmpty(error))
{
    throw new Exception(error);
}

if (output.Contains("okey"))
{
    return "okey"; // или можно вернуть output, если нужен весь вывод
}
else
{
    throw new Exception("Скрипт выполнился, но не вернул ожидаемый результат");
}
теперь мне надо модифицировать эти файлы.

1. создать новую функцию и запихнуть весь код туда
2. вместо sys.argv (считывание из командной строки) мне нужно передавать параметры
3. в конце функция должна возвращать тот результат, что тебе нужен

Сравни, что поменялось, чтобы лучше понять

переделанный cerenada.py
Python:
import requests
import search_element
from lxml import html, etree
import csv
import sys

def func(URL):
    if 'cerenada.ru' not in URL: raise Exception('wrong url')

    response = requests.get(URL)
    response.raise_for_status()

    XPATH_CARDS = "//ul[contains(@class, 'products')]/li"

    XPATH_TITLE = ".//h2/text()"
    XPATH_LINK = ".//a[contains(@href, 'product')]/@href"
    XPATH_IMAGE = ".//img/@src"  # коллекция

    source = html.fromstring(response.text)


    cards = search_element.find_elements(source, XPATH_CARDS)

    result = []
    for card in cards:
        mass = []
        title_card = search_element.find_element_in_element(card, XPATH_TITLE)
        link_card = search_element.find_element_in_element(card, XPATH_LINK)

        images_card_el = search_element.find_elements_in_element(card, XPATH_IMAGE)
        images_card = ':'.join(images_card_el)
        mass.append(title_card)
        mass.append(link_card)
        mass.append(images_card)
        result.append(mass)

    with open('result.csv', 'w', encoding='utf-8-sig', newline='') as file:
        writer = csv.writer(file, delimiter=';')
        writer.writerows(result)

    print("okey")

    # добавляем ещё что вернём в конце
    return "okey"
переделанный upper.py
Python:
def test(phrase):
    result = phrase.upper()

    # добавляем ещё что вернём в конце
    return result
теперь можем разворачивать API. Создаем новый файл new_api.py. благодаря тому, что мы запихали весь код в функции, мы сможем их импортировать. остальная информация есть в статье

new_api.py
Python:
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from cerenada import func  # импортировали ф-ю func из файла cerenada.py
from upper import test   # аналогично из файла upper.py

app = FastAPI()

@app.get("/parsing/", response_class=PlainTextResponse)
async def upper_text(url: str):
    return func(url)

# вот эту часть меняем, остальное просто копируем
@app.get("/upper/", response_class=PlainTextResponse)
async def upper_text(name: str):
    return test(name)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
134426

==
134427


Единственное, если у тебя многопоток, то возможно и не стоит "переезжать", я в эту сторону ещё подробно не смотрел. как вариант хотя бы делать запрос не из кубика, а из c# и лочить его, чтобы несколько потоков не залезли.
 
Последнее редактирование:
  • Спасибо
Реакции: Asmus003

Asmus003

Client
Регистрация
25.03.2018
Сообщения
297
Благодарностей
68
Баллы
28
там нужно модифицировать скрипты, а для этого нужно соображать, что именно они делают. поэтому я и попросил прикрепить их сюда. попробую объяснить, хотя твой случай может быть иным.

сейчас я создал 2 файла .py (2 скрипта), которые работают через "запуск программы".

так, первый файл "cerenada.py"
Python:
import requests
import search_element
from lxml import html, etree
import csv
import sys

if len(sys.argv) != 2: raise Exception('Не верно переданы аргументы')
URL = sys.argv[1]
print(URL)
if 'cerenada.ru' not in URL: raise Exception('wrong url')

response = requests.get(URL)
response.raise_for_status()

XPATH_CARDS = "//ul[contains(@class, 'products')]/li"

XPATH_TITLE = ".//h2/text()"
XPATH_LINK = ".//a[contains(@href, 'product')]/@href"
XPATH_IMAGE = ".//img/@src"  # коллекция

source = html.fromstring(response.text)


cards = search_element.find_elements(source, XPATH_CARDS)

result = []
for card in cards:
    mass = []
    title_card = search_element.find_element_in_element(card, XPATH_TITLE)
    link_card = search_element.find_element_in_element(card, XPATH_LINK)

    images_card_el = search_element.find_elements_in_element(card, XPATH_IMAGE)
    images_card = ':'.join(images_card_el)
    mass.append(title_card)
    mass.append(link_card)
    mass.append(images_card)
    result.append(mass)

with open('result.csv', 'w', encoding='utf-8-sig', newline='') as file:
    writer = csv.writer(file, delimiter=';')
    writer.writerows(result)

print("okey")
2-й файл upper.py
Python:
import sys

if len(sys.argv) != 2: raise Exception('Не верно переданы аргументы')
phrase = sys.argv[1]

result = phrase.upper()
sys.stdout.write(result)
sys.stdout.flush()
пример как он запускается на текущий момент (аналог кубика запуск программы):
C#:
System.Diagnostics.Process cmd = new System.Diagnostics.Process();
cmd.StartInfo.FileName = "cmd.exe";
cmd.StartInfo.RedirectStandardInput = true;
cmd.StartInfo.RedirectStandardOutput = true;
cmd.StartInfo.RedirectStandardError = true; // Добавляем перехват stderr
cmd.StartInfo.CreateNoWindow = true;
cmd.StartInfo.StandardOutputEncoding = Encoding.GetEncoding(866);
cmd.StartInfo.StandardErrorEncoding = Encoding.GetEncoding(866); // Кодировка для stderr
cmd.StartInfo.UseShellExecute = false;
cmd.Start();

cmd.StandardInput.WriteLine(String.Format("cd " + project.Variables["py_path"].Value));
cmd.StandardInput.WriteLine(String.Format("python cerenada.py " + project.Variables["arg"].Value));
cmd.StandardInput.Flush();
cmd.StandardInput.Close();

// Читаем вывод и ошибки
string output = cmd.StandardOutput.ReadToEnd();
string error = cmd.StandardError.ReadToEnd();

cmd.WaitForExit();

// Если есть ошибки - выбрасываем исключение
if (!string.IsNullOrEmpty(error))
{
    throw new Exception(error);
}

if (output.Contains("okey"))
{
    return "okey"; // или можно вернуть output, если нужен весь вывод
}
else
{
    throw new Exception("Скрипт выполнился, но не вернул ожидаемый результат");
}
теперь мне надо модифицировать эти файлы.

1. создать новую функцию и запихнуть весь код туда
2. вместо sys.argv (считывание из командной строки) мне нужно передавать параметры
3. в конце функция должна возвращать тот результат, что тебе нужен

Сравни, что поменялось, чтобы лучше понять

переделанный cerenada.py
Python:
import requests
import search_element
from lxml import html, etree
import csv
import sys

def func(URL):
    if 'cerenada.ru' not in URL: raise Exception('wrong url')

    response = requests.get(URL)
    response.raise_for_status()

    XPATH_CARDS = "//ul[contains(@class, 'products')]/li"

    XPATH_TITLE = ".//h2/text()"
    XPATH_LINK = ".//a[contains(@href, 'product')]/@href"
    XPATH_IMAGE = ".//img/@src"  # коллекция

    source = html.fromstring(response.text)


    cards = search_element.find_elements(source, XPATH_CARDS)

    result = []
    for card in cards:
        mass = []
        title_card = search_element.find_element_in_element(card, XPATH_TITLE)
        link_card = search_element.find_element_in_element(card, XPATH_LINK)

        images_card_el = search_element.find_elements_in_element(card, XPATH_IMAGE)
        images_card = ':'.join(images_card_el)
        mass.append(title_card)
        mass.append(link_card)
        mass.append(images_card)
        result.append(mass)

    with open('result.csv', 'w', encoding='utf-8-sig', newline='') as file:
        writer = csv.writer(file, delimiter=';')
        writer.writerows(result)

    print("okey")

    # добавляем ещё что вернём в конце
    return "okey"
переделанный upper.py
Python:
def test(phrase):
    result = phrase.upper()

    # добавляем ещё что вернём в конце
    return result
теперь можем разворачивать API. Создаем новый файл new_api.py. благодаря тому, что мы запихали весь код в функции, мы сможем их импортировать. остальная информация есть в статье

new_api.py
Python:
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from cerenada import func  # импортировали ф-ю func из файла cerenada.py
from upper import test   # аналогично из файла upper.py

app = FastAPI()

@app.get("/parsing/", response_class=PlainTextResponse)
async def upper_text(url: str):
    return func(url)

# вот эту часть меняем, остальное просто копируем
@app.get("/upper/", response_class=PlainTextResponse)
async def upper_text(name: str):
    return test(name)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
Посмотреть вложение 134426
==
Посмотреть вложение 134427

Единственное, если у тебя многопоток, то возможно и не стоит "переезжать", я в эту сторону ещё подробно не смотрел. как вариант хотя бы делать запрос не из кубика, а из c# и лочить его, чтобы несколько потоков не залезли.
спасибо большое за развернутый ответ!
я сейчас даже не с командной строкой работаю, а с батниками. потому что ошибка в командной строке выходит кракозябрами всегда.
 
  • Спасибо
Реакции: volody00

Lest

Client
Регистрация
20.03.2020
Сообщения
90
Благодарностей
89
Баллы
18
Спасибо за статью, интересный способ, возьму его в практику, если в шаблоне будет много раз вызываться разные методы пайтона. Хотя тогда лучше уже всё чисто на нём написать)
 

Кто просматривает тему: (Всего: 1, Пользователи: 0, Гости: 1)