18 промтов для GitHub Copilot: от коммитов до code review

# Напиши функцию на Python для парсинга JSON-логов.
# Функция должна:
# - принимать строку лога
# - возвращать словарь с полями: level, message, timestamp
# - обрабатывать исключения (если строка не JSON — вернуть None)
# - логировать ошибки через logging
# Используй typing для аннотаций.

import json
import logging
from typing import Optional, Dict, Any
from datetime import datetime

logger = logging.getLogger(__name__)

def parse_json_log(log_line: str) -> Optional[Dict[str, Any]]:
    """
    Parse a JSON log line into a structured dictionary.

    Args:
        log_line: A string containing a JSON log entry.

    Returns:
        A dictionary with keys 'level', 'message', 'timestamp', or None if parsing fails.
    """
    try:
        data = json.loads(log_line)
        return {
            "level": data.get("level", "INFO"),
            "message": data.get("message", ""),
            "timestamp": datetime.fromisoformat(data.get("timestamp", "1970-01-01T00:00:00"))
        }
    except (json.JSONDecodeError, ValueError, TypeError) as e:
        logger.error(f"Failed to parse log line: {e}")
        return None

# Напиши тесты на pytest для функции parse_json_log.
# Тесты должны покрывать:
# - корректный JSON с level, message, timestamp
# - некорректный JSON (ошибка парсинга)
# - отсутствие поля timestamp (должен вернуть дефолтную дату)
# - пустая строка
# Используй parametrize.

import pytest
from datetime import datetime
from your_module import parse_json_log

@pytest.mark.parametrize("log_line, expected_level, expected_message", [
    ('{"level": "ERROR", "message": "Timeout", "timestamp": "2026-07-23T10:00:00"}', "ERROR", "Timeout"),
    ('{"level": "INFO", "message": "Start"}', "INFO", "Start"),
    ('', None, None),
    ('not json', None, None),
])
def test_parse_json_log(log_line, expected_level, expected_message):
    result = parse_json_log(log_line)
    if expected_level is None:
        assert result is None
    else:
        assert result["level"] == expected_level
        assert result["message"] == expected_message

# Сгенерируй сообщение коммита для следующих изменений:
# - добавлена функция parse_json_log
# - добавлены тесты для неё
# - обновлён requirements.txt (добавлен pytest)
# Сообщение должно быть в формате conventional commits: feat

|fix|chore: краткое описание

feat: add parse_json_log function with error handling and tests

- Implement parse_json_log to safely parse JSON log lines
- Add pytest tests covering valid, invalid, and empty inputs
- Add pytest to project dependencies

# Напиши docstring в стиле Google для следующего класса:
class DatabaseConnection:
    def __init__(self, host: str, port: int, timeout: int = 30):
        self.host = host
        self.port = port
        self.timeout = timeout
    def connect(self):
        pass
    def query(self, sql: str):
        pass

class DatabaseConnection:
    """
    Manages a connection to a database server.

    Attributes:
        host: Database server hostname or IP address.
        port: Port number for the database connection.
        timeout: Connection timeout in seconds (default: 30).

    Methods:
        connect(): Establish the connection to the database.
        query(sql): Execute a SQL query and return results.
    """
    def __init__(self, host: str, port: int, timeout: int = 30):
        self.host = host
        self.port = port
        self.timeout = timeout

    def connect(self):
        """Establish connection to the database server."""
        pass

    def query(self, sql: str):
        """
        Execute a SQL query.

        Args:
            sql: The SQL query string to execute.

        Returns:
            Query result as a list of dictionaries.
        """
        pass

# Объясни, что делает следующий код, строка за строкой:
def process_data(items):
    result = []
    for item in items:
        if item.get('type') == 'A':
            value = item['value'] * 2
        elif item.get('type') == 'B':
            value = item['value'] ** 2
        else:
            value = item['value']
        result.append({'id': item['id'], 'computed': value})
    return result

Функция process_data принимает список словарей items. Для каждого элемента:
- Если type равен 'A', удваивает value.
- Если type равен 'B', возводит value в квадрат.
- Иначе оставляет value без изменений.
Возвращает список словарей с полями id и computed.

# Отрефактори этот код, используя словарь для маппинга типов:
def process_data(items):
    result = []
    for item in items:
        if item.get('type') == 'A':
            value = item['value'] * 2
        elif item.get('type') == 'B':
            value = item['value'] ** 2
        else:
            value = item['value']
        result.append({'id': item['id'], 'computed': value})
    return result

def process_data(items):
    type_map = {
        'A': lambda v: v * 2,
        'B': lambda v: v ** 2,
    }
    result = []
    for item in items:
        transformer = type_map.get(item.get('type'), lambda v: v)
        value = transformer(item['value'])
        result.append({'id': item['id'], 'computed': value})
    return result

# Напиши SQL-запрос для PostgreSQL:
# Таблицы: users (id, name, email), orders (id, user_id, amount, created_at)
# Нужно: для каждого пользователя показать общую сумму заказов за последние 30 дней
# Результат: name, email, total_amount
# Отсортировать по убыванию суммы

SELECT u.name, u.email, COALESCE(SUM(o.amount), 0) AS total_amount
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
    AND o.created_at >= NOW() - INTERVAL '30 days'
GROUP BY u.id, u.name, u.email
ORDER BY total_amount DESC;

# Перепиши эту функцию на Go:
def parse_json_log(log_line):
    import json
    try:
        data = json.loads(log_line)
        return {"level": data.get("level", "INFO"), "message": data.get("message", "")}
    except json.JSONDecodeError:
        return None

package main

import (
    "encoding/json"
    "log"
)

type LogEntry struct {
    Level   string `json:"level"`
    Message string `json:"message"`
}

func ParseJSONLog(logLine string) *LogEntry {
    var entry LogEntry
    err := json.Unmarshal([]byte(logLine), &entry)
    if err != nil {
        log.Printf("Failed to parse log line: %v", err)
        return nil
    }
    if entry.Level == "" {
        entry.Level = "INFO"
    }
    return &entry
}

# Оптимизируй этот код для обработки больших списков (миллионы элементов):
def find_duplicates(items):
    duplicates = []
    for i in range(len(items)):
        for j in range(i+1, len(items)):
            if items[i] == items[j]:
                duplicates.append(items[i])
    return duplicates

def find_duplicates(items):
    seen = set()
    duplicates = set()
    for item in items:
        if item in seen:
            duplicates.add(item)
        else:
            seen.add(item)
    return list(duplicates)

# Проведи code review следующего кода. Найди:
# - потенциальные баги
# - проблемы безопасности
# - нарушения стиля (PEP8)
# - предложения по улучшению

async def fetch_data(url):
    import requests
    response = requests.get(url)
    return response.json()

## Code Review

### Баги:
- Использование синхронного requests в асинхронной функции. Нужно заменить на aiohttp или httpx.

### Безопасность:
- Нет обработки таймаута. При медленном сервере функция зависнет.
- Нет проверки статуса ответа. Если сервер вернёт 404 или 500, json() вызовет исключение.

### Стиль:
- Импорт requests внутри функции — лучше вынести в начало файла.

### Предложения:
- Добавить параметр timeout.
- Обработать исключения (aiohttp.ClientError, asyncio.TimeoutError).
- Использовать typing: async def fetch_data(url: str) -> dict.

# Найди уязвимости безопасности в этом коде. Особенно SQL-инъекции и XSS:
def get_user(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    cursor.execute(query)
    return cursor.fetchone()

## Безопасность

### SQL-инъекция:
- Прямая подстановка user_id в строку запроса. Если user_id содержит '; DROP TABLE users; --', это может уничтожить таблицу.
- Решение: использовать параметризованные запросы:
  cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))

### Рекомендации:
- Валидировать тип user_id (должен быть int).
- Использовать ORM (например, SQLAlchemy) для автоматической защиты.

# Напиши Dockerfile для Python-приложения с зависимостями из requirements.txt.
# Используй официальный образ Python 3.12 slim.
# Установи зависимости, скопируй код, укажи команду запуска: python app.py
# Добавь healthcheck.

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

HEALTHCHECK --interval=30s --timeout=3s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

CMD ["python", "app.py"]

# Напиши GitHub Actions workflow для Python-проекта:
# - триггер: push в main, pull request
# - шаги: checkout, setup Python 3.12, install deps, run tests (pytest), lint (flake8)
# - используй матрицу для Python 3.11 и 3.12

name: CI

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12"]

    steps:
    - uses: actions/checkout@v4
    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v5
      with:
        python-version: ${{ matrix.python-version }}
    - name: Install dependencies
      run: pip install -r requirements.txt
    - name: Lint with flake8
      run: flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
    - name: Test with pytest
      run: pytest

# Сгенерируй OpenAPI 3.0 спецификацию для REST API:
# - POST /users: создать пользователя (name, email)
# - GET /users/{id}: получить пользователя по ID
# - DELETE /users/{id}: удалить пользователя
# Все ответы в JSON.

openapi: "3.0.0"
info:
  title: User API
  version: "1.0.0"
paths:
  /users:
    post:
      summary: Create a new user
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              properties:
                name:
                  type: string
                email:
                  type: string
      responses:
        '201':
          description: User created
  /users/{id}:
    get:
      summary: Get user by ID
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: integer
      responses:
        '200':
          description: User details
    delete:
      summary: Delete user
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: integer
      responses:
        '204':
          description: User deleted

# Отрефактори этот код, используя паттерн Strategy для обработки разных типов платежей:
def process_payment(payment_type, amount):
    if payment_type == "credit_card":
        # logic for credit card
        return amount * 1.02
    elif payment_type == "paypal":
        # logic for paypal
        return amount * 1.03
    elif payment_type == "crypto":
        # logic for crypto
        return amount * 1.01

from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
    @abstractmethod
    def calculate_fee(self, amount: float) -> float:
        pass

class CreditCardPayment(PaymentStrategy):
    def calculate_fee(self, amount: float) -> float:
        return amount * 1.02

class PayPalPayment(PaymentStrategy):
    def calculate_fee(self, amount: float) -> float:
        return amount * 1.03

class CryptoPayment(PaymentStrategy):
    def calculate_fee(self, amount: float) -> float:
        return amount * 1.01

def process_payment(strategy: PaymentStrategy, amount: float) -> float:
    return strategy.calculate_fee(amount)

# Напиши Python-скрипт для миграции пользователей из CSV в PostgreSQL.
# CSV имеет колонки: id, name, email, created_at
# Таблица: users (id SERIAL, name TEXT, email TEXT UNIQUE, created_at TIMESTAMP)
# Используй psycopg2 и обработку ошибок.

import csv
import psycopg2
from psycopg2.extras import execute_values

def migrate_users(csv_path, db_config):
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()

    with open(csv_path, 'r') as f:
        reader = csv.DictReader(f)
        users = [(row['name'], row['email'], row['created_at']) for row in reader]

    try:
        execute_values(
            cur,
            "INSERT INTO users (name, email, created_at) VALUES %s ON CONFLICT (email) DO NOTHING",
            users
        )
        conn.commit()
        print(f"Migrated {len(users)} users")
    except Exception as e:
        conn.rollback()
        print(f"Migration failed: {e}")
    finally:
        cur.close()
        conn.close()

# Напиши Python-скрипт, который читает лог-файл в реальном времени (tail -f)
# и подсчитывает количество ошибок (строк с 'ERROR') за последние 5 минут.
# Используй asyncio и aiofiles.

import asyncio
import aiofiles
from collections import deque
from datetime import datetime, timedelta

async def monitor_log(file_path):
    error_times = deque()
    async with aiofiles.open(file_path, 'r') as f:
        await f.seek(0, 2)  # go to end
        while True:
            line = await f.readline()
            if line:
                if 'ERROR' in line:
                    error_times.append(datetime.now())
                    # Remove errors older than 5 minutes
                    cutoff = datetime.now() - timedelta(minutes=5)
                    while error_times and error_times[0] < cutoff:
                        error_times.popleft()
                    print(f"Errors in last 5 min: {len(error_times)}")
            else:
                await asyncio.sleep(0.1)

# Напиши Makefile для Python-проекта:
# - install: установка зависимостей
# - test: запуск тестов
# - lint: проверка flake8
# - clean: удаление __pycache__ и .pyc
# - docker-build: сборка Docker-образа

.PHONY: install test lint clean docker-build

install:
    pip install -r requirements.txt

test:
    pytest -v --cov=.

lint:
    flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics

clean:
    find . -type d -name "__pycache__" -exec rm -rf {} +
    find . -type f -name "*.pyc" -delete

docker-build:
    docker build -t myapp:latest .