marc
/
cgn-appointments
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: marc:main
Branches Tags
marc:main
marc:1.1.0
marc:1.0.3
marc:1.0.2
marc:1.0.1
marc:1.0.0
..
compare: marc:1.0.0
Branches Tags
marc:main
marc:1.1.0
marc:1.0.3
marc:1.0.2
marc:1.0.1
marc:1.0.0

No commits in common. "main" and "1.0.0" have entirely different histories.
main ... 1.0.0

6 changed files with 19 additions and 245 deletions
Show Stats Expand all files Collapse all files

1
.idea/cgn-appointments.iml
View File

@ -4,7 +4,6 @@
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
<excludeFolder url="file://$MODULE_DIR$/.venv" />
<excludeFolder url="file://$MODULE_DIR$/dist" />
</content>
<orderEntry type="jdk" jdkName="Python 3.12 (.venv)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />

185
README.md
View File

@ -1,186 +1 @@
# cgn-appointments
This programm allows you to scrape the appointments from the website of the city
of Cologne (Stadt Köln) and send a notification via ntfy if a new appointment is
available.
## Installation
A good way to install `cgn-appointments` is via [pipx](https://pipx.pypa.io/stable/).
```bash
pipx install --index-url https://git.pub.solar/api/packages/marc/pypi/simple/ --pip-args="--extra-index-url https://pypi.org/simple" cgn-appointments
```
## Configuration
Execute `cgn-appointments` for the first time to create the configuration file.
Then edit the configuration file and adapt it to your needs.
```yaml
---
# If this url does not work anymore, visit https://www.stadt-koeln.de/service/produkte/00416/index.html,
# copy the url of the "Termin Buchen" tile and replace the url below.
url: 'https://termine.stadt-koeln.de/m/kundenzentren/extern/calendar/?uid=b5a5a394-ec33-4130-9af3-490f99517071&wsid=e570a1ea-7b3d-43f6-bf43-3e60b3d7d888&lang=de&set_lang_ui=de&rev=rfOtF#top'
# Which services should be checked for free appointments?
# Copy the exact name of the service from the website.
services:
- 'Personalausweis - Antrag'
- 'Reisepass - Antrag (seit 01.01.2024 auch für Kinder unter 12 Jahren)'
# In which locations should the services be checked?
# Use the name as displayed on the website.
locations:
- Ehrenfeld
- Kalk
# Max time between today and a new appointment to notify about the new
# appointment. Set to -1 to notify about all new appointments.
max_timedelta: 14
# Path to the CSV file to store the scraped appointments
# csv_path: ~/Termine.csv
# Name of the CSV file to store the scraped appointments
csv_name: 'appointments.csv'
# Regex to extract the date from the website
# There should be no need to change this
date_regex: '(\\d{2}\\.\\d{2}\\.\\d{4}\\s\\d{2}:\\d{2})'
# Date format to store the date in the CSV file (should match the date_regex)
# There should be no need to change this
date_format: '%d.%m.%Y %H:%M'
# ntfy configuration
# See https://ntfy.sh/ for more information
# Choose a topic name that is unique to you
ntfy:
server: https://ntfy.sh/
topic: public_cgn_appintments_83e0c8db1f51a7044b6431ddb2814c11
title: 'A new appointment is available!'
message: 'A new appointment is available in %s: %s'
tags:
- tada
priority: 3 # 1-5
# Configure logging
# Advanced users can change the logging configuration here
# The loglevel can be set to DEBUG, INFO, WARNING, ERROR, CRITICAL
logging:
version: 1
disable_existing_loggers: false
formatters:
simple:
format: '[%(levelname)s|%(module)s|L%(lineno)d] %(asctime)s: %(message)s'
datefmt: '%Y-%m-%dT%H:%M:%S%z'
json:
fmt_keys:
level: levelname
message: message
timestamp: timestamp
logger: name
module: module
function: funcName
line: lineno
thread_name: threadName
handlers:
stderr:
class: logging.StreamHandler
formatter: simple
stream: ext://sys.stderr
level: DEBUG
file:
class: logging.handlers.RotatingFileHandler
formatter: json
level: INFO
maxBytes: 10000000
backupCount: 3
queue_handler:
class: logging.handlers.QueueHandler
handlers:
- stderr
- file
respect_handler_level: true
loggers:
root:
handlers:
- queue_handler
level: DEBUG
```
## Usage
You can use `cg-appointments` with only the configuration file or with
additional command line arguments which will override the values set in the
configuration file.
Using the `--services` flag, make sure to use `"double quotes"` around the
service name if it contains spaces.
```bash
usage: cgn-appointments [-h] [-s SERVICES [SERVICES ...]]
[-l LOCATIONS [LOCATIONS ...]] [-t MAX_TIMEDELTA]
[--config-file CONFIG_FILE] [--csv-file CSV_FILE]
[--log-file LOG_FILE]
[--log-level {CRITICAL,ERROR,WARNING,INFO,DEBUG,NOTSET}]
Scrapes appointments from termine.stadt-koeln.de an sends a message to a ntfy
server.
options:
-h, --help show this help message and exit
-s SERVICES [SERVICES ...], --services SERVICES [SERVICES ...]
Services to check
-l LOCATIONS [LOCATIONS ...], --locations LOCATIONS [LOCATIONS ...]
Locations to check
-t MAX_TIMEDELTA, --max-timedelta MAX_TIMEDELTA
Maximum timedelta in days to notify about new
appointments
--config-file CONFIG_FILE
Path to the configuration file
--csv-file CSV_FILE Path to the csv file, which stores the last fetched
dates
--log-file LOG_FILE Path to logfile
--log-level {CRITICAL,ERROR,WARNING,INFO,DEBUG,NOTSET}
Logging Level
```
### Example
```bash
cgn-apppointments \
--services "Personalausweis - Antrag" "Reisepass - Antrag (seit 01.01.2024 auch für Kinder unter 12 Jahren)" \
--locations Ehrenfeld Kalk \
--max-timedelta 7 \
--config-file /path/to/config.yaml \
--csv-file /path/to/csvfile.csv \
--log-file /path/to/logfile.log \
--log-level INFO
```
## Scheduled Execution via `cron`
On linux systems you can use `cron` to schedule the execution of
`cgn-appointments`.
Find the path to the `cgn-appointments` executable
```bash
which cgn-appointments
```
Open the crontab file
```bash
crontab -e
```
Add the following line to the crontab file to execute `cgn-appointments` every
30 minutes
```bash
*/30 * * * * /path/to/cgn-appointments > /dev/null 2>&1
```

2
pyproject.toml
View File

@ -7,7 +7,7 @@ name = "cgn-appointments"
dynamic = ["version"]
description = 'Scrapes appointments from termine.stadt-koeln.de an sends a message to a ntfy server.'
readme = "README.md"
requires-python = ">=3.12"
requires-python = ">=3.8"
license = "MIT"
keywords = ["selenim", "cologne", "scraper"]
authors = [

2
src/cgnappointments/__about__.py
View File

@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: 2024-present Marc Koch <marc-koch@posteo.de>
#
# SPDX-License-Identifier: MIT
__version__ = "1.1.0"
__version__ = "1.0.0"

68
src/cgnappointments/__main__.py
View File

@ -50,14 +50,6 @@ def parse_arguments() -> dict:
help="Locations to check",
required=False,
)
argparser.add_argument(
"-t",
"--max-timedelta",
action="store",
type=int,
help="Maximum timedelta in days to notify about new appointments",
required=False,
)
argparser.add_argument(
"--config-file",
action="store",
@ -65,13 +57,6 @@ def parse_arguments() -> dict:
help="Path to the configuration file",
required=False,
)
argparser.add_argument(
"--csv-file",
action="store",
type=Path,
help="Path to the csv file, which stores the last fetched dates",
required=False,
)
argparser.add_argument(
"--log-file",
action="store",
@ -100,8 +85,7 @@ def update_config_with_args(config: dict, args: dict) -> dict:
update_config = {
"services": args.get("services"),
"locations": args.get("locations"),
"max_timedelta": args.get("max_timedelta"),
"csv_path": args.get("csv_file"),
"csv_path": args.get("csv_path"),
}
for key, value in update_config.items():
if value is not None:
@ -155,21 +139,16 @@ def define_csv_path(csv_path: str|None, csv_name: str|None) -> Path:
csv_path = Path(csv_path) if csv_path else None
csv_name = Path(csv_name) if csv_name else None
if csv_path is not None and csv_path.is_dir() and csv_name is not None:
if csv_path is not None and csv_path.is_file():
return csv_path
elif csv_path is not None and csv_path.is_dir() and csv_name is not None:
return csv_path / csv_name
elif csv_path is not None and csv_path.is_dir() and csv_name is None:
return csv_path / "cgn-appointments.csv"
elif csv_path is not None:
csv_path.touch()
return csv_path
elif csv_name is not None:
csv_path = Path(user_data_dir()) / csv_name
csv_path.touch()
return csv_path
return Path(user_data_dir()) / csv_name
else:
csv_path = Path(user_data_dir()) / "cgn-appointments"
csv_path.touch()
return csv_path / "cgn-appointments.csv"
return Path(user_data_dir()) / "cgn-appointments.csv"
def select_options(services, selects):
@ -266,7 +245,6 @@ def main():
url = config.get("url")
services = config.get("services")
check_locations = config.get("locations")
max_timedelta = config.get("max_timedelta")
csv_name = config.get("csv_name", "cgn-appointments.csv")
csv_path = define_csv_path(config.get("csv_path"), csv_name)
date_regex = config.get("date_regex")
@ -392,31 +370,17 @@ def main():
if new_date and new_date != previous_date:
logger.info(f"New appointment found for {name}: {new_date}",
extra={"location": name, "previous_date": previous_date,
"new_date": new_date, "max_timedelta": max_timedelta})
"new_date": new_date})
lines.append((name, new_date.strftime(date_format)))
# Send notification if new date is within timedelta or
# timedelta is not set
time_delta = (new_date - datetime.now()).days if max_timedelta > 0 else False
if time_delta == False or time_delta <= max_timedelta:
logger.info(f"Sending notification for new appointment.",
extra={"location": name, "new_date": new_date,
"time_delta": time_delta,
"max_timedelta": max_timedelta})
ntfy(
ntfy_server,
ntfy_topic,
ntfy_title,
ntfy_message % (name, new_date),
session_url,
ntfy_tags,
ntfy_priority,
)
else:
logger.info(f"New appointment is not within timedelta.",
extra={"location": name, "new_date": new_date,
"time_delta": time_delta,
"max_timedelta": max_timedelta})
ntfy(
ntfy_server,
ntfy_topic,
ntfy_title,
ntfy_message % (name, new_date),
session_url,
ntfy_tags,
ntfy_priority,
)
elif previous_date is not None:
lines.append((name, previous_date.strftime(date_format)))

6
src/cgnappointments/config_template.yaml
View File

@ -11,10 +11,6 @@ locations:
- Ehrenfeld
- Kalk
# Max time between today and a new appointment to notify about the new
# appointment. Set to -1 to notify about all new appointments.
max_timedelta: 14
# Path to the CSV file to store the scraped appointments
# csv_path: ~/Termine.csv
@ -22,7 +18,7 @@ max_timedelta: 14
csv_name: 'appointments.csv'
# Regex to extract the date from the website
date_regex: '(\d{2}\.\d{2}\.\d{4}\s\d{2}:\d{2})'
date_regex: '(\\d{2}\\.\\d{2}\\.\\d{4}\\s\\d{2}:\\d{2})'
# Date format to store the date in the CSV file (should match the date_regex)
date_format: '%d.%m.%Y %H:%M'