Coverage for app_modules/cameras_api.py: 88%

1""" 

2cameras_api.py 

3 

4Purpose: 

5 JSON APIs for camera CRUD and metadata retrieval used by the selection page. 

6 

7Routes: 

8 - GET /api/kamere 

9 - POST /api/kamere/add 

10 - POST /api/kamere/rename 

11 - POST /api/kamere/delete 

12""" 

13 

14from flask import Blueprint, jsonify, request, session 

15import sqlite3 

16from .db import get_db 

17from .security import login_required 

18from .paths import STATIC_PATH 

19from .images_service import latest_from_db_or_fs 

20from .helpers import format_dt 

21from .assets import _static_exists 

22from .security_enhancements import validate_api_input 

23 

24 

25bp = Blueprint('cameras_api', __name__) 

26 

27 

28@bp.route('/api/kamere') 

29@login_required 

30def api_cameras(): 

31 db = get_db() 

32 # dynamic fields 

33 cols = [row[1] for row in db.execute('PRAGMA table_info(cameras)').fetchall()] 

34 has_model = 'model' in set(cols) 

35 select_fields = 'camera_id, camera_name, file_paths' + (', model' if has_model else '') 

36 cur = db.execute(f'SELECT {select_fields} FROM cameras WHERE user_id = ? ORDER BY camera_name COLLATE NOCASE', (session['user_id'],)) 

37 rows = [] 

38 for r in cur.fetchall(): 

39 cam_id = str(r['camera_id']) 

40 latest_dt, _ = latest_from_db_or_fs(cam_id, r['file_paths'] or '', STATIC_PATH) 

41 thumb_url = '/static/camera_render.png' if _static_exists('camera_render.png') else '/static/webicon180x180.png' 

42 model = (r['model'] if has_model else None) or 'Vision mini' 

43 rows.append({ 

44 'camera_id': cam_id, 

45 'camera_name': r['camera_name'], 

46 'last_active': format_dt(latest_dt) if latest_dt else 'Nema aktivnosti', 

47 'thumbnail_url': thumb_url, 

48 'model': model 

49 }) 

50 resp = jsonify({'cameras': rows}) 

51 resp.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0' 

52 resp.headers['Pragma'] = 'no-cache' 

53 return resp 

54 

55 

56@bp.route('/api/kamere/add', methods=['POST']) 

57@login_required 

58def api_add_camera(): 

59 data = request.get_json(silent=True) or {} 

60 

61 # Enhanced input validation 

62 schema = { 

63 'camera_id': {'type': str, 'max_length': 12, 'required': True}, 

64 'camera_name': {'type': str, 'max_length': 100, 'required': True} 

65 } 

66 valid, error_msg, validated_data = validate_api_input(data, schema) 

67 if not valid: 

68 return jsonify({'success': False, 'message': error_msg}), 400 

69 

70 camera_id = validated_data['camera_id'].strip() 

71 camera_name = validated_data['camera_name'].strip() 

72 

73 if not (camera_id.isdigit() and len(camera_id) == 12): 

74 return jsonify({'success': False, 'message': 'Broj kamere mora imati točno 12 znamenki.'}), 400 

75 if not camera_name: 

76 return jsonify({'success': False, 'message': 'Ime kamere je obavezno.'}), 400 

77 db = get_db() 

78 exists = db.execute('SELECT 1 FROM cameras WHERE user_id=? AND camera_id=?', (session['user_id'], camera_id)).fetchone() 

79 if exists: 

80 return jsonify({'success': False, 'message': 'Kamera već postoji.'}), 409 

81 # Insert 

82 cols = [row[1] for row in db.execute('PRAGMA table_info(cameras)').fetchall()] 

83 has_model = 'model' in set(cols) 

84 try: 

85 if has_model: 

86 db.execute('INSERT INTO cameras (user_id, camera_id, camera_name, model) VALUES (?, ?, ?, ?)', (session['user_id'], camera_id, camera_name, 'Vision mini')) 

87 else: 

88 db.execute('INSERT INTO cameras (user_id, camera_id, camera_name) VALUES (?, ?, ?)', (session['user_id'], camera_id, camera_name)) 

89 db.commit() 

90 return jsonify({'success': True}) 

91 except sqlite3.Error: 

92 return jsonify({'success': False, 'message': 'Greška pri spremanju.'}), 500 

93 

94 

95@bp.route('/api/kamere/rename', methods=['POST']) 

96@login_required 

97def api_rename_camera(): 

98 data = request.get_json(silent=True) or {} 

99 camera_id = str(data.get('camera_id', '')).strip() 

100 new_name = (data.get('camera_name') or '').strip() 

101 if not (camera_id.isdigit() and len(camera_id) == 12): 

102 return jsonify({'success': False, 'message': 'Neispravan ID kamere.'}), 400 

103 if not new_name: 

104 return jsonify({'success': False, 'message': 'Ime je obavezno.'}), 400 

105 db = get_db() 

106 try: 

107 cur = db.execute('UPDATE cameras SET camera_name=? WHERE user_id=? AND camera_id=?', (new_name, session['user_id'], camera_id)) 

108 db.commit() 

109 if cur.rowcount == 0: 

110 return jsonify({'success': False, 'message': 'Kamera nije pronađena.'}), 404 

111 return jsonify({'success': True}) 

112 except sqlite3.Error: 

113 return jsonify({'success': False, 'message': 'Greška baze.'}), 500 

114 

115 

116@bp.route('/api/kamere/delete', methods=['POST']) 

117@login_required 

118def api_delete_camera(): 

119 data = request.get_json(silent=True) or {} 

120 camera_id = str(data.get('camera_id', '')).strip() 

121 if not (camera_id.isdigit() and len(camera_id) == 12): 

122 return jsonify({'success': False, 'message': 'Neispravan ID kamere.'}), 400 

123 db = get_db() 

124 try: 

125 cur = db.execute('DELETE FROM cameras WHERE user_id=? AND camera_id=?', (session['user_id'], camera_id)) 

126 db.commit() 

127 if cur.rowcount == 0: 

128 return jsonify({'success': False, 'message': 'Kamera nije pronađena.'}), 404 

129 return jsonify({'success': True}) 

130 except sqlite3.Error: 

131 return jsonify({'success': False, 'message': 'Greška baze.'}), 500 

132 

133