psf / black

The uncompromising Python code formatter
https://black.readthedocs.io/en/stable/
MIT License
38.82k stars 2.45k forks source link

Black won't format file as it produces invalid code #4439

Closed RedX2501 closed 2 months ago

RedX2501 commented 2 months ago

Describe the bug

I installed the VS extensions and tried to format a file.

In the log output it said it can't format the file and told me to report the bug.

To Reproduce

For example, take this code:

from typing import List
import requests
import json
import sys
from datetime import datetime, timedelta, time
from zoneinfo import ZoneInfo
from typing import NamedTuple
import locale

class Subcalendar(NamedTuple):
    name: str
    id: int

class TimeSlot(NamedTuple):
    start: datetime
    end: datetime

    def covered(this, event):
        event_start = datetime.strptime(event['start_dt'], '%Y-%m-%dT%H:%M:%S%z')
        event_end = datetime.strptime(event['end_dt'], '%Y-%m-%dT%H:%M:%S%z')
        is_covered = this.start >= event_start and this.end <= event_end
        return is_covered

def load_config(config_files):
    config = {}
    for config_file in config_files:
        try:
            with open(config_file, 'r') as file:
                config_content = json.load(file)
            config.update(config_content)
        except Exception as e:
            print(f"Error loading configuration {config_file}: {e}")
            sys.exit(1)

    return config

def fetch_subcalendar_ids(api_key, calendar_id):
    headers = {"Teamup-Token": api_key}
    url = f"https://api.teamup.com/{calendar_id}/subcalendars"
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    subcalendars = response.json().get('subcalendars', [])
    return {sub['name']: sub['id'] for sub in subcalendars}

def fetch_events(api_key: str, calendar_id: str, start_date: datetime, end_date: datetime, sub_calendars: List[Subcalendar]):
    subcalendar_query = [('subcalendarId[]', sub_calendar.id) for sub_calendar in sub_calendars]
    headers = {"Teamup-Token": api_key}
    params = [
        ('startDate', start_date.strftime('%Y-%m-%d')),
        ('endDate', end_date.strftime('%Y-%m-%d')),
        *subcalendar_query
    ]
    url = f"https://api.teamup.com/{calendar_id}/events"
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get('events', [])

def create_available_slots(start_datetime: datetime, end_datetime: datetime) -> List[TimeSlot]:
    available_slots = []
    current_time = start_datetime
    time_inc = timedelta(minutes=15)
    while current_time < end_datetime:
        available_slots.append(TimeSlot(current_time, current_time + time_inc))
        current_time += time_inc
    return available_slots

def merge_slots(slots: List[TimeSlot]) -> List[TimeSlot]:
    new_slots = []
    if not slots:
        return new_slots

    current_slot = slots[0]._replace()
    for slot in slots[1:]:
        if current_slot.end == slot.start:
            current_slot = current_slot._replace(end=slot.end)
        else:
            new_slots.append(current_slot)
            current_slot = slot._replace()

    new_slots.append(current_slot)

    return new_slots    

def get_free_time_slots(events, date: datetime, start_time:time, end_time:time) -> List[TimeSlot]:
    start_datetime = date.replace(hour=start_time.hour, minute=start_time.minute, second=start_time.second, microsecond=0)
    end_datetime = date.replace(hour=end_time.hour, minute=end_time.minute, second=end_time.second, microsecond=0)

    available_slots = create_available_slots(start_datetime, end_datetime)

    for event in events:
        available_slots = [available_slot for available_slot in available_slots if not available_slot.covered(event)]
        if not available_slots:
            break

    return merge_slots(available_slots)

def send_telegram_message(bot_token, channels, message):
    for channel in channels:
        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
        payload = {            'chat_id': channel['id'],            'text': message,            'format': 'MarkdownV2'
            }
        response = requests.post(url, data=payload)
        response.raise_for_status()
    print("Message sent successfully!")

def fetch_subcalendar_id_from_name(config) -> List[Subcalendar]:
    subcalendar_ids = fetch_subcalendar_ids(config['teamup_api_key'], config['calendar_id'])
    interesting_calendars = config['subcalendars_to_check']
    subcalendars_to_check = [
        Subcalendar(name, int(subcalendar_ids[name])) for name in interesting_calendars if name in subcalendar_ids
    ]    

    calendar_not_found = False
    for name in interesting_calendars:
        if not name in subcalendar_ids:
            print(f"Calendar {name} not found in response.")
            calendar_not_found = True

    if calendar_not_found:
        print(f"Known calendars: {subcalendar_ids}")
        sys.exit(1)

    return subcalendars_to_check

def convert_to_date(zone, text, days_to_check=None) -> datetime:
    date = datetime.strptime(text, "%Y-%m-%d") if text else None

    if date:
        date = date.replace(tzinfo=zone)
    else:
        date = datetime.now(zone)
        if days_to_check:
             date += timedelta(days=days_to_check)

    return date

def parse_time(text: str) -> time:
    return time.fromisoformat(text)

def check_slots_and_notify(config: map, dry_run: bool = False) -> None:

    sub_calendars = fetch_subcalendar_id_from_name(config)

    tzone = ZoneInfo(config["timezone"])
    start_date = convert_to_date(tzone, config.get('start_date')).replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
    end_date = convert_to_date(tzone, config.get('end_date'), config.get('days_to_check', 7)).replace(hour=23, minute=59, second=59, microsecond=999)

    print(f"Checking Thekendienst between ({start_date:%A}) {start_date:%Y-%m-%d} and ({end_date:%A}) {end_date:%Y-%m-%d}.")

    events = fetch_events(
        config['teamup_api_key'], 
        config['calendar_id'], 
        start_date, 
        end_date,
        sub_calendars
    )

    print(f"Found {len(events)} events in the time range for calendars '{"', '".join(sub_calendar.name for sub_calendar in sub_calendars)}'.")

    message = ""
    for i in range((end_date - start_date).days + 1):
        check_date = start_date + timedelta(days=i)
        day_of_week = check_date.strftime('%A')  # Get the day name, e.g., 'Monday'

        if day_of_week in config['time_slots']:
            start_time = parse_time(config['time_slots'][day_of_week]['start'])
            end_time = parse_time(config['time_slots'][day_of_week]['end'])

            free_time_slots = get_free_time_slots(events, check_date, start_time, end_time)

            # kind of a hack but I don't want to install any packages
            old_locale = locale.getlocale()
            locale.setlocale(locale.LC_ALL, "de_DE.utf8")

            if free_time_slots:
                free_slots = ' und '.join([f"{slot.start:%H:%M} bis {slot.end:%H:%M}" for slot in free_time_slots])
                message += (
                    f"🚨 {free_time_slots[0].start:%A} den {free_time_slots[0].start:%d.%b.} von {free_slots}. 🚨\n"
                )
            locale.setlocale(locale.LC_ALL, old_locale)

    if message:
        message = config['header'] + message + "\n" + config['footer']
    else:
        message = config['no_open_slots']

    if dry_run:
        print("Messsage that would be sent on Telegram:")
        print(message)
    else:
        send_telegram_message(config['telegram_bot_token'], config['telegram_channels'], message)

if __name__ == "__main__":
    default_config_file = 'reportMissingThekendienstConfig.json'
    dry_run = "--dry-run" in sys.argv
    config_files = [arg for arg in sys.argv[1:] if arg != '--dry-run']
    if not config_files:
        config_files = [default_config_file]
    config = load_config(config_files)
    check_slots_and_notify(config, dry_run=dry_run)

I don't know the arguments the vs code extension runs it with.

This is the contents of the log file:

  File "/home/gui/.vscode/extensions/ms-python.black-formatter-2024.2.0/bundled/libs/black/__init__.py", line 1536, in assert_equivalent
    dst_ast = parse_ast(dst)
              ^^^^^^^^^^^^^^
  File "/home/gui/.vscode/extensions/ms-python.black-formatter-2024.2.0/bundled/libs/black/parsing.py", line 148, in parse_ast
    raise SyntaxError(first_error)
from typing import List
import requests
import json
import sys
from datetime import datetime, timedelta, time
from zoneinfo import ZoneInfo
from typing import NamedTuple
import locale

class Subcalendar(NamedTuple):
    name: str
    id: int

class TimeSlot(NamedTuple):
    start: datetime
    end: datetime

    def covered(this, event):
        event_start = datetime.strptime(event["start_dt"], "%Y-%m-%dT%H:%M:%S%z")
        event_end = datetime.strptime(event["end_dt"], "%Y-%m-%dT%H:%M:%S%z")
        is_covered = this.start >= event_start and this.end <= event_end
        return is_covered

def load_config(config_files):
    config = {}
    for config_file in config_files:
        try:
            with open(config_file, "r") as file:
                config_content = json.load(file)
            config.update(config_content)
        except Exception as e:
            print(f"Error loading configuration {config_file}: {e}")
            sys.exit(1)

    return config

def fetch_subcalendar_ids(api_key, calendar_id):
    headers = {"Teamup-Token": api_key}
    url = f"https://api.teamup.com/{calendar_id}/subcalendars"
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    subcalendars = response.json().get("subcalendars", [])
    return {sub["name"]: sub["id"] for sub in subcalendars}

def fetch_events(
    api_key: str,
    calendar_id: str,
    start_date: datetime,
    end_date: datetime,
    sub_calendars: List[Subcalendar],
):
    subcalendar_query = [
        ("subcalendarId[]", sub_calendar.id) for sub_calendar in sub_calendars
    ]
    headers = {"Teamup-Token": api_key}
    params = [
        ("startDate", start_date.strftime("%Y-%m-%d")),
        ("endDate", end_date.strftime("%Y-%m-%d")),
        *subcalendar_query,
    ]
    url = f"https://api.teamup.com/{calendar_id}/events"
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get("events", [])

def create_available_slots(
    start_datetime: datetime, end_datetime: datetime
) -> List[TimeSlot]:
    available_slots = []
    current_time = start_datetime
    time_inc = timedelta(minutes=15)
    while current_time < end_datetime:
        available_slots.append(TimeSlot(current_time, current_time + time_inc))
        current_time += time_inc
    return available_slots

def merge_slots(slots: List[TimeSlot]) -> List[TimeSlot]:
    new_slots = []
    if not slots:
        return new_slots

    current_slot = slots[0]._replace()
    for slot in slots[1:]:
        if current_slot.end == slot.start:
            current_slot = current_slot._replace(end=slot.end)
        else:
            new_slots.append(current_slot)
            current_slot = slot._replace()

    new_slots.append(current_slot)

    return new_slots

def get_free_time_slots(
    events, date: datetime, start_time: time, end_time: time
) -> List[TimeSlot]:
    start_datetime = date.replace(
        hour=start_time.hour,
        minute=start_time.minute,
        second=start_time.second,
        microsecond=0,
    )
    end_datetime = date.replace(
        hour=end_time.hour,
        minute=end_time.minute,
        second=end_time.second,
        microsecond=0,
    )

    available_slots = create_available_slots(start_datetime, end_datetime)

    for event in events:
        available_slots = [
            available_slot
            for available_slot in available_slots
            if not available_slot.covered(event)
        ]
        if not available_slots:
            break

    return merge_slots(available_slots)

def send_telegram_message(bot_token, channels, message):
    for channel in channels:
        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
        payload = {"chat_id": channel["id"], "text": message, "format": "MarkdownV2"}
        response = requests.post(url, data=payload)
        response.raise_for_status()
    print("Message sent successfully!")

def fetch_subcalendar_id_from_name(config) -> List[Subcalendar]:
    subcalendar_ids = fetch_subcalendar_ids(
        config["teamup_api_key"], config["calendar_id"]
    )
    interesting_calendars = config["subcalendars_to_check"]
    subcalendars_to_check = [
        Subcalendar(name, int(subcalendar_ids[name]))
        for name in interesting_calendars
        if name in subcalendar_ids
    ]

    calendar_not_found = False
    for name in interesting_calendars:
        if not name in subcalendar_ids:
            print(f"Calendar {name} not found in response.")
            calendar_not_found = True

    if calendar_not_found:
        print(f"Known calendars: {subcalendar_ids}")
        sys.exit(1)

    return subcalendars_to_check

def convert_to_date(zone, text, days_to_check=None) -> datetime:
    date = datetime.strptime(text, "%Y-%m-%d") if text else None

    if date:
        date = date.replace(tzinfo=zone)
    else:
        date = datetime.now(zone)
        if days_to_check:
            date += timedelta(days=days_to_check)

    return date

def parse_time(text: str) -> time:
    return time.fromisoformat(text)

def check_slots_and_notify(config: map, dry_run: bool = False) -> None:

    sub_calendars = fetch_subcalendar_id_from_name(config)

    tzone = ZoneInfo(config["timezone"])
    start_date = convert_to_date(tzone, config.get("start_date")).replace(
        hour=0, minute=0, second=0, microsecond=0
    ) + timedelta(days=1)
    end_date = convert_to_date(
        tzone, config.get("end_date"), config.get("days_to_check", 7)
    ).replace(hour=23, minute=59, second=59, microsecond=999)

    print(
        f"Checking Thekendienst between ({start_date:%A}) {start_date:%Y-%m-%d} and ({end_date:%A}) {end_date:%Y-%m-%d}."
    )

    events = fetch_events(
        config["teamup_api_key"],
        config["calendar_id"],
        start_date,
        end_date,
        sub_calendars,
    )

    print(
        f"Found {len(events)} events in the time range for calendars '{"
        ", "
        ".join(sub_calendar.name for sub_calendar in sub_calendars)}'."
    )

    message = ""
    for i in range((end_date - start_date).days + 1):
        check_date = start_date + timedelta(days=i)
        day_of_week = check_date.strftime("%A")  # Get the day name, e.g., 'Monday'

        if day_of_week in config["time_slots"]:
            start_time = parse_time(config["time_slots"][day_of_week]["start"])
            end_time = parse_time(config["time_slots"][day_of_week]["end"])

            free_time_slots = get_free_time_slots(
                events, check_date, start_time, end_time
            )

            # kind of a hack but I don't want to install any packages
            old_locale = locale.getlocale()
            locale.setlocale(locale.LC_ALL, "de_DE.utf8")

            if free_time_slots:
                free_slots = " und ".join(
                    [
                        f"{slot.start:%H:%M} bis {slot.end:%H:%M}"
                        for slot in free_time_slots
                    ]
                )
                message += f"🚨 {free_time_slots[0].start:%A} den {free_time_slots[0].start:%d.%b.} von {free_slots}. 🚨\n"
            locale.setlocale(locale.LC_ALL, old_locale)

    if message:
        message = config["header"] + message + "\n" + config["footer"]
    else:
        message = config["no_open_slots"]

    if dry_run:
        print("Messsage that would be sent on Telegram:")
        print(message)
    else:
        send_telegram_message(
            config["telegram_bot_token"], config["telegram_channels"], message
        )

if __name__ == "__main__":
    default_config_file = "reportMissingThekendienstConfig.json"
    dry_run = "--dry-run" in sys.argv
    config_files = [arg for arg in sys.argv[1:] if arg != "--dry-run"]
    if not config_files:
        config_files = [default_config_file]
    config = load_config(config_files)
    check_slots_and_notify(config, dry_run=dry_run)

Expected behavior

The file is formatted.

Environment

Linux Ubuntu VSCode

Version: 1.92.2
Commit: fee1edb8d6d72a0ddff41e5f71a671c23ed924b9
Date: 2024-08-14T17:29:30.058Z
Electron: 30.1.2
ElectronBuildId: 9870757
Chromium: 124.0.6367.243
Node.js: 20.14.0
V8: 12.4.254.20-electron.0
OS: Linux x64 6.8.0-40-generic

Python Formatter

Name: Black Formatter
Id: ms-python.black-formatter
Description: Formatting support for Python files using the Black formatter.
Version: 2024.2.0
Publisher: Microsoft
VS Marketplace Link: https://marketplace.visualstudio.com/items?itemName=ms-python.black-formatter
hauntsaninja commented 2 months ago

I think this is related to nested quotes in f strings and is fixed in newer Black (e.g. I can't repro a crash with Black 24.4.2)

JelleZijlstra commented 2 months ago

The issue is this line:

    print(f"Found {len(events)} events in the time range for calendars '{"', '".join(sub_calendar.name for sub_calendar in sub_calendars)}'.")

This uses syntax that was only added in Python 3.12 (PEP 701), and as @hauntsaninja said, recent versions of Black support this.

RedX2501 commented 2 months ago

Yeah, that was it. Thank you.