cnicholas
/
unifi-access-airbnb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor visitor creation and improve error handling 

- Simplify create_visitor method to match successful API calls
- Enhance error logging for better debugging
- Update README with new usage instructions
- Add CHANGELOG to track project changes
main
Keith Herrington 2024-09-11 19:59:20 -07:00
parent cc7ba33e48
commit a17df9c330
5 changed files with 177 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
CHANGELOG.md Normal file
View File

@ -0,0 +1,23 @@
# Changelog
All notable changes to this project will be documented in this file.
## [Unreleased]
## [0.2.0] - 2024-09-11
### Added
- Automatic setting of default door group when only one is available
- Improved error logging for API interactions
- CHANGELOG file to track project changes
### Changed
- Refactored `create_visitor` method to simplify API calls
- Updated `process_reservations` method to use new `create_visitor` implementation
- Improved error handling in API request methods
### Fixed
- Issue with visitor creation failing due to invalid parameters
### Updated
- README with more detailed usage instructions and configuration details

56
README.md
View File

@ -19,71 +19,41 @@ This project integrates UniFi Access with Airbnb reservations, automating the pr
## Installation
1. Clone the repository:
git clone https://github.com/keithah/unifi-access-airbnb.git
git clone https://github.com/yourusername/unifi-access-airbnb.git
cd unifi-access-airbnb
3. Install the required packages:
2. Install the required packages:
pip install -r requirements.txt
4. Copy the example configuration file and edit it with your settings:
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
python3 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)
- `-l [LOG_FILE]` or `--log [LOG_FILE]`: Specify a log file
- `--list-door-groups`: List available door groups
## Scheduling
## Configuration
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
Edit the `unifi.conf` file with your specific settings. Key sections include:
- `[UniFi]`: UniFi Access API settings
- `[Hostex]`: Hostex API settings (if used)
- `[Airbnb]`: Airbnb ICS feed URL (if used)
- `[Door]`: Default door group ID for visitor access
- `[Visitor]`: Check-in and check-out times
## 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.

17
config.py
View File

@ -1,9 +1,21 @@
import configparser
import logging
logger = logging.getLogger(__name__)
def load_config():
config = configparser.ConfigParser()
config.read('unifi.conf')
logger.debug("Loaded sections from unifi.conf: %s", config.sections())
if 'Door' not in config:
logger.error("'Door' section not found in unifi.conf")
elif 'default_group_id' not in config['Door']:
logger.error("'default_group_id' not found in 'Door' section of unifi.conf")
else:
logger.debug("Found default_group_id in config: %s", config['Door']['default_group_id'])
return {
'api_host': config['UniFi']['api_host'],
'api_token': config['UniFi']['api_token'],
@ -13,7 +25,7 @@ def load_config():
'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'],
'default_door_group_id': config['Door'].get('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'],
@ -21,3 +33,6 @@ def load_config():
'log_file': config['General']['log_file'],
'pin_code_digits': int(config['General']['pin_code_digits'])
}
logger.debug("Loaded configuration: %s", {k: v for k, v in config.items() if k != 'api_token'})
return config

35
main.py
View File

@ -15,28 +15,49 @@ 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")
parser.add_argument('--list-door-groups', action='store_true', help="List available door groups")
args = parser.parse_args()
config = load_config()
log_file = args.log or config['log_file']
# Initialize logging first
logger = logging.getLogger(__name__)
log_file = args.log or 'unifi_access.log' # Default log file if not specified
logger = setup_logging(args.verbose, log_file)
try:
config = load_config()
logger.debug(f"Loaded config: {config}")
except Exception as e:
logger.error(f"Error loading configuration: {str(e)}")
return
try:
unifi_manager = UnifiAccessManager(config)
except ValueError as e:
logger.error(f"Error initializing UnifiAccessManager: {str(e)}")
return
if args.list_door_groups:
unifi_manager.print_door_groups()
return
hostex_manager = HostexManager(config)
ics_parser = ICSParser(config)
notification_manager = NotificationManager(config)
try:
logger.debug("Script started")
logger.info("Script started")
if config['use_hostex']:
logger.info("Fetching reservations from Hostex")
reservations = hostex_manager.fetch_reservations()
elif config['use_ics']:
logger.info("Parsing ICS file")
reservations = ics_parser.parse_ics()
else:
logger.error("No valid reservation source configured")
return
logger.info(f"Processing {len(reservations)} reservations")
unifi_manager.process_reservations(reservations)
summary = unifi_manager.generate_summary()
@ -47,15 +68,15 @@ def main():
if config['simplepush_enabled'] and unifi_manager.has_changes():
notification_manager.send_notification("UniFi Access Update", summary)
logger.debug("Simplepush notification sent")
logger.info("Simplepush notification sent")
else:
logger.debug("No Simplepush notification sent (no changes or Simplepush not enabled)")
logger.info("No Simplepush notification sent (no changes or Simplepush not enabled)")
logger.debug("Script completed successfully")
logger.info("Script completed successfully")
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
finally:
logger.debug("Script execution finished")
logger.info("Script execution finished")
if __name__ == "__main__":
main()

129
unifi_access.py
View File

@ -7,57 +7,76 @@ 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.default_door_group_id = config.get('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):
self.logger.debug(f"Loaded default_door_group_id from config: {self.default_door_group_id}")
if not self.default_door_group_id:
self.logger.warning("default_door_group_id is not set in config. Attempting to fetch available door groups.")
self.set_default_door_group()
else:
self.logger.debug(f"Initialized with default_door_group_id: {self.default_door_group_id}")
def set_default_door_group(self):
door_groups = self.fetch_door_groups()
if len(door_groups) == 1:
self.default_door_group_id = door_groups[0]['id']
self.logger.info(f"Automatically set default_door_group_id to the only available group: {self.default_door_group_id}")
elif len(door_groups) > 1:
self.logger.error("Multiple door groups available. Please specify default_group_id in unifi.conf")
raise ValueError("Multiple door groups available. Please specify default_group_id in unifi.conf")
else:
self.logger.error("No door groups available")
raise ValueError("No door groups available")
def create_visitor(self, first_name, last_name, 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"}
{
"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}")
self.logger.debug(f"Creating visitor with data: {json.dumps(data, indent=2)}")
try:
response = requests.post(url, json=data, headers=headers, verify=False)
self.logger.debug(f"API response status code: {response.status_code}")
self.logger.debug(f"API response content: {response.text}")
response.raise_for_status() # Raise an exception for bad status codes
response_data = response.json()
if response_data.get('code') == 'SUCCESS':
visitor_id = response_data.get('data', {}).get('id')
if visitor_id:
self.logger.debug(f"Created visitor with ID: {visitor_id}")
return True
else:
self.logger.error("Visitor ID not found in the response")
return False
else:
return True
self.logger.error(f"API returned an error: {response_data.get('msg')}")
return False
except requests.exceptions.RequestException as e:
self.logger.error(f"Request failed: {str(e)}")
return False
def fetch_visitors(self):
url = f"{self.api_host}/api/v1/developer/visitors"
@ -66,8 +85,15 @@ class UnifiAccessManager:
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers, verify=False)
self.logger.debug(f"Fetch visitors API response status code: {response.status_code}")
if response.status_code == 200:
return response.json()["data"]
data = response.json()
if 'data' in data:
return data['data']
else:
self.logger.error(f"Unexpected response format: {data}")
return []
else:
self.logger.error("Failed to fetch existing visitors")
return []
@ -81,10 +107,11 @@ class UnifiAccessManager:
params = {"is_force": "true"} if is_completed else {}
response = requests.delete(url, headers=headers, params=params, verify=False)
self.logger.debug(f"Delete visitor API response status code: {response.status_code}")
if response.status_code != 200:
self.logger.error(f"Failed to delete visitor account: {visitor_id}")
return False
else:
return True
def process_reservations(self, reservations):
@ -92,19 +119,20 @@ class UnifiAccessManager:
next_month = today + datetime.timedelta(days=30)
existing_visitors = self.fetch_visitors()
self.logger.debug(f"Processing {len(reservations)} reservations")
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 ""
first_name, last_name = guest_name.split(" ", 1) if " " in guest_name else (guest_name, "")
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),
datetime.datetime.fromtimestamp(int(v["start_time"])).date() == check_in_date and
datetime.datetime.fromtimestamp(int(v["end_time"])).date() == check_out_date),
None
)
@ -116,14 +144,14 @@ class UnifiAccessManager:
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)
success = self.create_visitor(first_name, last_name, 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()
visitor_end = datetime.datetime.fromtimestamp(int(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)
@ -146,3 +174,32 @@ class UnifiAccessManager:
def has_changes(self):
return bool(self.changes['added'] or self.changes['deleted'])
def fetch_door_groups(self):
url = f"{self.api_host}/api/v1/developer/door_groups"
headers = {
"Authorization": f"Bearer {self.api_token}",
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers, verify=False)
self.logger.debug(f"Fetch door groups API response status code: {response.status_code}")
if response.status_code == 200:
data = response.json()
if 'data' in data:
return data['data']
else:
self.logger.error(f"Unexpected response format: {data}")
return []
else:
self.logger.error("Failed to fetch door groups")
return []
def print_door_groups(self):
door_groups = self.fetch_door_groups()
if door_groups:
print("Available Door Groups:")
for group in door_groups:
print(f"ID: {group['id']}, Name: {group['name']}")
else:
print("No door groups found or failed to fetch door groups.")