Initial commit
Keith Herrington
parent
bf4d7a75d1
commit
4e4e3a7f5a
|
|
@ -0,0 +1,7 @@
|
|||
unifi.conf
|
||||
*.log
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
# UniFi Access Airbnb Integration
|
||||
|
||||
This project integrates UniFi Access with Airbnb reservations, automating the process of creating and managing visitor access for your Airbnb guests.
|
||||
|
||||
## Features
|
||||
|
||||
- Fetch reservations from Hostex API or Airbnb ICS feed
|
||||
- Create UniFi Access visitor accounts for upcoming guests
|
||||
- Assign PIN codes to visitors based on their phone number
|
||||
- Automatically delete past or completed visitor accounts
|
||||
- Send notifications via Simplepush for updates and failures
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Python 3.7+
|
||||
- UniFi Access system
|
||||
- Airbnb account with ICS feed URL or Hostex API access
|
||||
|
||||
## Installation
|
||||
|
||||
1. Clone the repository:
|
||||
git clone https://github.com/yourusername/unifi-access-airbnb.git
|
||||
cd unifi-access-airbnb
|
||||
|
||||
2. Install the required packages:
|
||||
pip install -r requirements.txt
|
||||
|
||||
3. Copy the example configuration file and edit it with your settings:
|
||||
cp unifi.conf.example unifi.conf
|
||||
nano unifi.conf
|
||||
|
||||
## Configuration
|
||||
|
||||
Edit the `unifi.conf` file with your specific settings. Here's an explanation of each section:
|
||||
|
||||
- `[UniFi]`: UniFi Access API settings
|
||||
- `[Hostex]`: Hostex API settings (if used)
|
||||
- `[Airbnb]`: Airbnb ICS feed URL (if used)
|
||||
- `[Simplepush]`: Simplepush notification settings
|
||||
- `[Door]`: Default door group ID for visitor access
|
||||
- `[Visitor]`: Check-in and check-out times
|
||||
- `[General]`: General settings like log file name and PIN code length
|
||||
|
||||
## Usage
|
||||
|
||||
Run the script using:
|
||||
python main.py
|
||||
|
||||
Optional arguments:
|
||||
- `-v` or `--verbose`: Increase output verbosity
|
||||
- `-l [LOG_FILE]` or `--log [LOG_FILE]`: Specify a log file (default is set in the config file)
|
||||
|
||||
## Scheduling
|
||||
|
||||
To run the script automatically, you can set up a cron job. For example, to run it every hour:
|
||||
|
||||
1. Open your crontab file:
|
||||
crontab -e
|
||||
2. Add the following line (adjust the path as needed):
|
||||
0 * * * * /usr/bin/python3 /path/to/unifi-access-airbnb/main.py
|
||||
|
||||
## Contributing
|
||||
|
||||
Contributions are welcome! Please feel free to submit a Pull Request.
|
||||
|
||||
1. Fork the project
|
||||
2. Create your feature branch (`git checkout -b feature/AmazingFeature`)
|
||||
3. Commit your changes (`git commit -m 'Add some AmazingFeature'`)
|
||||
4. Push to the branch (`git push origin feature/AmazingFeature`)
|
||||
5. Open a Pull Request
|
||||
|
||||
## License
|
||||
|
||||
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
|
||||
|
||||
## Acknowledgments
|
||||
|
||||
- UniFi Access for their API
|
||||
- Hostex for their reservation management API
|
||||
- Airbnb for providing ICS feed functionality
|
||||
- Simplepush for their notification service
|
||||
|
||||
## Support
|
||||
|
||||
If you encounter any problems or have any questions, please open an issue on the GitHub repository.
|
||||
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
import configparser
|
||||
|
||||
def load_config():
|
||||
config = configparser.ConfigParser()
|
||||
config.read('unifi.conf')
|
||||
|
||||
return {
|
||||
'api_host': config['UniFi']['api_host'],
|
||||
'api_token': config['UniFi']['api_token'],
|
||||
'hostex_api_url': config['Hostex']['api_url'],
|
||||
'hostex_api_key': config['Hostex']['api_key'],
|
||||
'ics_url': config.get('Airbnb', 'ics_url', fallback=None),
|
||||
'simplepush_enabled': config['Simplepush'].getboolean('enabled', fallback=False),
|
||||
'simplepush_key': config['Simplepush'].get('key', fallback=None),
|
||||
'simplepush_url': config['Simplepush'].get('url', fallback=None),
|
||||
'default_door_group_id': config['Door']['default_group_id'],
|
||||
'check_in_time': config['Visitor']['check_in_time'],
|
||||
'check_out_time': config['Visitor']['check_out_time'],
|
||||
'use_hostex': 'Hostex' in config and config['Hostex']['api_key'],
|
||||
'use_ics': config.get('Airbnb', 'ics_url', fallback=None) is not None,
|
||||
'log_file': config['General']['log_file'],
|
||||
'pin_code_digits': int(config['General']['pin_code_digits'])
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
import requests
|
||||
import logging
|
||||
|
||||
class HostexManager:
|
||||
def __init__(self, config):
|
||||
self.api_url = config['hostex_api_url']
|
||||
self.api_key = config['hostex_api_key']
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def fetch_reservations(self):
|
||||
url = f"{self.api_url}/reservations"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
response = requests.get(url, headers=headers)
|
||||
if response.status_code == 200:
|
||||
reservations = response.json()["data"]["reservations"]
|
||||
self.logger.debug(f"Fetched {len(reservations)} reservations from Hostex")
|
||||
return reservations
|
||||
else:
|
||||
self.logger.error(f"Failed to fetch reservations from Hostex. Status code: {response.status_code}")
|
||||
return []
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
import requests
|
||||
import icalendar
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
class ICSParser:
|
||||
def __init__(self, config):
|
||||
self.ics_url = config['ics_url']
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def parse_ics(self):
|
||||
response = requests.get(self.ics_url)
|
||||
cal = icalendar.Calendar.from_ical(response.text)
|
||||
reservations = []
|
||||
for event in cal.walk("VEVENT"):
|
||||
start = event.get("DTSTART").dt
|
||||
end = event.get("DTEND").dt
|
||||
description = event.get("DESCRIPTION", "")
|
||||
if not description:
|
||||
self.logger.debug(f"Skipping event with start date {start.date()} due to missing description")
|
||||
continue
|
||||
pin_code = ""
|
||||
for line in description.split("\n"):
|
||||
if line.startswith("Phone Number (Last 4 Digits):"):
|
||||
pin_code = line.split(": ")[1].strip()
|
||||
break
|
||||
reservations.append({
|
||||
"check_in_date": start.date() if isinstance(start, datetime.datetime) else start,
|
||||
"check_out_date": end.date() if isinstance(end, datetime.datetime) else end,
|
||||
"guests": [{"name": "Airbnb Guest", "phone": pin_code}],
|
||||
"status": "accepted"
|
||||
})
|
||||
self.logger.debug(f"Parsed {len(reservations)} reservations from ICS file")
|
||||
return reservations
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
import argparse
|
||||
import logging
|
||||
from config import load_config
|
||||
from unifi_access import UnifiAccessManager
|
||||
from hostex_api import HostexManager
|
||||
from ics_parser import ICSParser
|
||||
from notification import NotificationManager
|
||||
from utils import setup_logging
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="UniFi Access Visitor Management")
|
||||
parser.add_argument('-v', '--verbose', action='store_true', help="Increase output verbosity")
|
||||
parser.add_argument('-l', '--log', help="Log output to file")
|
||||
args = parser.parse_args()
|
||||
|
||||
config = load_config()
|
||||
log_file = args.log or config['log_file']
|
||||
logger = setup_logging(args.verbose, log_file)
|
||||
|
||||
unifi_manager = UnifiAccessManager(config)
|
||||
hostex_manager = HostexManager(config)
|
||||
ics_parser = ICSParser(config)
|
||||
notification_manager = NotificationManager(config)
|
||||
|
||||
try:
|
||||
logger.debug("Script started")
|
||||
|
||||
if config['use_hostex']:
|
||||
reservations = hostex_manager.fetch_reservations()
|
||||
elif config['use_ics']:
|
||||
reservations = ics_parser.parse_ics()
|
||||
else:
|
||||
logger.error("No valid reservation source configured")
|
||||
return
|
||||
|
||||
unifi_manager.process_reservations(reservations)
|
||||
|
||||
summary = unifi_manager.generate_summary()
|
||||
logger.info(summary)
|
||||
|
||||
if config['simplepush_enabled'] and unifi_manager.has_changes():
|
||||
notification_manager.send_notification("UniFi Access Update", summary)
|
||||
|
||||
logger.debug("Script completed successfully")
|
||||
except Exception as e:
|
||||
logger.error(f"An error occurred: {str(e)}", exc_info=True)
|
||||
finally:
|
||||
logger.debug("Script execution finished")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
import requests
|
||||
import logging
|
||||
|
||||
class NotificationManager:
|
||||
def __init__(self, config):
|
||||
self.enabled = config['simplepush_enabled']
|
||||
self.key = config['simplepush_key']
|
||||
self.url = config['simplepush_url']
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def send_notification(self, title, message, event="airbnb-access"):
|
||||
if not self.enabled:
|
||||
self.logger.debug("Simplepush is not enabled. Skipping notification.")
|
||||
return
|
||||
url = f"{self.url}/{self.key}/{title}/{message}/event/{event}"
|
||||
response = requests.get(url)
|
||||
if response.status_code != 200:
|
||||
self.logger.error("Failed to send Simplepush notification")
|
||||
else:
|
||||
self.logger.debug("Simplepush notification sent successfully")
|
||||
|
|
@ -0,0 +1,4 @@
|
|||
requests==2.26.0
|
||||
icalendar==4.0.7
|
||||
urllib3==1.26.7
|
||||
configparser==5.0.2
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
[UniFi]
|
||||
api_host = https://unifi-access-ip:12445
|
||||
api_token = your_unifi_access_api_token
|
||||
|
||||
[Hostex]
|
||||
api_url = https://api.hostex.io/v3
|
||||
api_key = your_hostex_api_key
|
||||
|
||||
[Airbnb]
|
||||
ics_url = https://www.airbnb.com/calendar/ical/your_ical_id.ics
|
||||
|
||||
[Simplepush]
|
||||
enabled = true
|
||||
key = your_simplepush_key
|
||||
url = https://api.simplepush.io/send
|
||||
|
||||
[Door]
|
||||
default_group_id = your_default_door_group_id
|
||||
|
||||
[Visitor]
|
||||
check_in_time = 16:00:00
|
||||
check_out_time = 11:00:00
|
||||
|
||||
[General]
|
||||
log_file = unifi_access_airbnb.log
|
||||
pin_code_digits = 4
|
||||
|
|
@ -0,0 +1,150 @@
|
|||
import requests
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
|
||||
class UnifiAccessManager:
|
||||
def __init__(self, config):
|
||||
self.api_host = config['api_host']
|
||||
self.api_token = config['api_token']
|
||||
self.default_door_group_id = config['default_door_group_id']
|
||||
self.check_in_time = datetime.time.fromisoformat(config['check_in_time'])
|
||||
self.check_out_time = datetime.time.fromisoformat(config['check_out_time'])
|
||||
self.pin_code_digits = config['pin_code_digits']
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self.changes = {'added': [], 'deleted': [], 'unchanged': []}
|
||||
|
||||
def create_visitor(self, first_name, last_name, remarks, phone_number, start_time, end_time):
|
||||
url = f"{self.api_host}/api/v1/developer/visitors"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
pin_code = phone_number[-self.pin_code_digits:] if phone_number and len(phone_number) >= self.pin_code_digits else ""
|
||||
data = {
|
||||
"first_name": first_name,
|
||||
"last_name": last_name,
|
||||
"remarks": remarks,
|
||||
"mobile_phone": phone_number,
|
||||
"email": "",
|
||||
"visitor_company": "",
|
||||
"start_time": start_time,
|
||||
"end_time": end_time,
|
||||
"visit_reason": "Other",
|
||||
"resources": [
|
||||
{"id": self.default_door_group_id, "type": "door_group"}
|
||||
]
|
||||
}
|
||||
response = requests.post(url, json=data, headers=headers, verify=False)
|
||||
if response.status_code != 200:
|
||||
self.logger.error(f"Failed to create visitor account for {first_name} {last_name}")
|
||||
return False
|
||||
else:
|
||||
visitor_id = response.json()["data"]["id"]
|
||||
return self.assign_pin_to_visitor(visitor_id, pin_code)
|
||||
|
||||
def assign_pin_to_visitor(self, visitor_id, pin_code):
|
||||
url = f"{self.api_host}/api/v1/developer/visitors/{visitor_id}/pin_codes"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
data = {
|
||||
"pin_code": pin_code
|
||||
}
|
||||
response = requests.put(url, json=data, headers=headers, verify=False)
|
||||
if response.status_code != 200:
|
||||
self.logger.error(f"Failed to assign PIN code to visitor: {visitor_id}")
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def fetch_visitors(self):
|
||||
url = f"{self.api_host}/api/v1/developer/visitors"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
response = requests.get(url, headers=headers, verify=False)
|
||||
if response.status_code == 200:
|
||||
return response.json()["data"]
|
||||
else:
|
||||
self.logger.error("Failed to fetch existing visitors")
|
||||
return []
|
||||
|
||||
def delete_visitor(self, visitor_id, is_completed=False):
|
||||
url = f"{self.api_host}/api/v1/developer/visitors/{visitor_id}"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
params = {"is_force": "true"} if is_completed else {}
|
||||
|
||||
response = requests.delete(url, headers=headers, params=params, verify=False)
|
||||
if response.status_code != 200:
|
||||
self.logger.error(f"Failed to delete visitor account: {visitor_id}")
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def is_pin_in_use(self, pin_code):
|
||||
visitors = self.fetch_visitors()
|
||||
for visitor in visitors:
|
||||
if "pin_codes" in visitor and pin_code in visitor["pin_codes"]:
|
||||
return True
|
||||
return False
|
||||
|
||||
def process_reservations(self, reservations):
|
||||
today = datetime.date.today()
|
||||
next_month = today + datetime.timedelta(days=30)
|
||||
existing_visitors = self.fetch_visitors()
|
||||
|
||||
for reservation in reservations:
|
||||
check_in_date = datetime.datetime.strptime(reservation["check_in_date"], "%Y-%m-%d").date()
|
||||
check_out_date = datetime.datetime.strptime(reservation["check_out_date"], "%Y-%m-%d").date()
|
||||
|
||||
if today <= check_in_date <= next_month and reservation["status"] == "accepted":
|
||||
guest_name = reservation["guests"][0]["name"] if reservation["guests"] else "Guest"
|
||||
remarks = json.dumps(reservation)
|
||||
phone_number = reservation["guests"][0].get("phone", "") if reservation["guests"] else ""
|
||||
|
||||
existing_visitor = next(
|
||||
(v for v in existing_visitors if
|
||||
datetime.datetime.fromtimestamp(v["start_time"]).date() == check_in_date and
|
||||
datetime.datetime.fromtimestamp(v["end_time"]).date() == check_out_date),
|
||||
None
|
||||
)
|
||||
|
||||
if existing_visitor:
|
||||
self.changes['unchanged'].append(f"{existing_visitor['first_name']} {existing_visitor['last_name']}")
|
||||
else:
|
||||
start_datetime = datetime.datetime.combine(check_in_date, self.check_in_time)
|
||||
end_datetime = datetime.datetime.combine(check_out_date, self.check_out_time)
|
||||
start_timestamp = int(start_datetime.timestamp())
|
||||
end_timestamp = int(end_datetime.timestamp())
|
||||
|
||||
success = self.create_visitor(guest_name, "", remarks, phone_number, start_timestamp, end_timestamp)
|
||||
if success:
|
||||
self.changes['added'].append(guest_name)
|
||||
else:
|
||||
self.logger.error(f"Failed to create visitor: {guest_name}")
|
||||
|
||||
for visitor in existing_visitors:
|
||||
visitor_end = datetime.datetime.fromtimestamp(visitor["end_time"]).date()
|
||||
is_completed = visitor.get("status") == "VISITED"
|
||||
if visitor_end < today or is_completed:
|
||||
success = self.delete_visitor(visitor["id"], is_completed)
|
||||
if success:
|
||||
self.changes['deleted'].append(f"{visitor['first_name']} {visitor['last_name']}")
|
||||
else:
|
||||
self.logger.error(f"Failed to delete visitor: {visitor['first_name']} {visitor['last_name']}")
|
||||
|
||||
def generate_summary(self):
|
||||
summary = "UniFi Access Update Summary:\n"
|
||||
summary += f"{len(self.changes['unchanged'])} existing visitors unchanged\n"
|
||||
summary += f"{len(self.changes['deleted'])} visitor(s) deleted\n"
|
||||
summary += f"{len(self.changes['added'])} visitor(s) added\n"
|
||||
return summary
|
||||
|
||||
def has_changes(self):
|
||||
return bool(self.changes['added'] or self.changes['deleted'])
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
import logging
|
||||
|
||||
def setup_logging(verbose, log_file):
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
|
||||
ch = logging.StreamHandler()
|
||||
ch.setLevel(logging.DEBUG if verbose else logging.INFO)
|
||||
ch.setFormatter(formatter)
|
||||
logger.addHandler(ch)
|
||||
|
||||
fh = logging.FileHandler(log_file)
|
||||
fh.setLevel(logging.DEBUG)
|
||||
fh.setFormatter(formatter)
|
||||
logger.addHandler(fh)
|
||||
|
||||
return logger
|
||||
Loading…
Reference in New Issue