Coverage for app_modules/map_routes.py: 82%

1""" 

2map_routes.py 

3 

4Purpose: 

5 Map and marker endpoints used by the Leaflet-based map UI. Provides logic for 

6 camera locations and generic fauna/structure markers. Performs schema 

7 migrations-on-boot for required columns/tables. 

8 

9Routes: 

10 - GET /karta 

11 - GET /get_available_cameras 

12 - POST /add_camera_marker 

13 - POST /add_item_marker 

14 - GET /get_marker_locations 

15 - POST /delete_marker 

16""" 

17 

18import sqlite3 

19from flask import Blueprint, render_template, jsonify, request, session 

20from .security import login_required 

21from .db import get_db 

22from .security_enhancements import validate_geographic_coordinates 

23 

24 

25bp = Blueprint('map_routes', __name__) 

26 

27 

28def _ensure_map_schema(db): 

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

30 if 'added_to_map' not in cols: 

31 try: 

32 db.execute('ALTER TABLE cameras ADD COLUMN added_to_map INTEGER DEFAULT 0') 

33 except sqlite3.Error: 

34 pass 

35 if 'location' not in cols: 

36 try: 

37 db.execute('ALTER TABLE cameras ADD COLUMN location TEXT') 

38 except sqlite3.Error: 

39 pass 

40 db.execute( 

41 'CREATE TABLE IF NOT EXISTS markers (' 

42 ' id INTEGER PRIMARY KEY AUTOINCREMENT,' 

43 ' user_id INTEGER NOT NULL,' 

44 ' type TEXT NOT NULL,' 

45 ' latitude REAL NOT NULL,' 

46 ' longitude REAL NOT NULL,' 

47 ' name TEXT)' 

48 ) 

49 db.commit() 

50 

51 

52@bp.route('/karta') 

53@login_required 

54def map_page(): 

55 return render_template('map.html') 

56 

57 

58@bp.route('/get_available_cameras') 

59@login_required 

60def get_available_cameras(): 

61 db = get_db() 

62 _ensure_map_schema(db) 

63 cur = db.execute( 

64 'SELECT camera_id, camera_name, IFNULL(added_to_map, 0) AS added_to_map' 

65 ' FROM cameras WHERE user_id=? AND IFNULL(added_to_map, 0)=0' 

66 ' ORDER BY camera_name COLLATE NOCASE', 

67 (session['user_id'],) 

68 ) 

69 data = [{'camera_id': row['camera_id'], 'name': row['camera_name'], 'added_to_map': row['added_to_map']} for row in cur.fetchall()] 

70 return jsonify(data) 

71 

72 

73@bp.route('/add_camera_marker', methods=['POST']) 

74@login_required 

75def add_camera_marker(): 

76 lat = request.form.get('lat') or request.json.get('lat') if request.is_json else None 

77 lng = request.form.get('lng') or request.json.get('lng') if request.is_json else None 

78 camera_id = (request.form.get('camera_id') or (request.json.get('camera_id') if request.is_json else '')).strip() 

79 if not lat or not lng or not camera_id: 

80 return jsonify({'success': False, 'message': 'Nedostaju podaci.'}), 400 

81 

82 # Use secure coordinate validation 

83 valid, error_msg, coordinates = validate_geographic_coordinates(lat, lng) 

84 if not valid: 

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

86 lat_f, lng_f = coordinates 

87 db = get_db() 

88 _ensure_map_schema(db) 

89 # Verify camera ownership 

90 row = db.execute('SELECT camera_name FROM cameras WHERE user_id=? AND camera_id=?', (session['user_id'], camera_id)).fetchone() 

91 if not row: 

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

93 location = f"{lat_f},{lng_f}" 

94 try: 

95 db.execute('UPDATE cameras SET location=?, added_to_map=1 WHERE user_id=? AND camera_id=?', (location, session['user_id'], camera_id)) 

96 db.commit() 

97 return jsonify({'success': True, 'camera_name': row['camera_name'], 'camera_id': camera_id}) 

98 except sqlite3.Error: 

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

100 

101 

102@bp.route('/add_item_marker', methods=['POST']) 

103@login_required 

104def add_item_marker(): 

105 lat = request.form.get('lat') or request.json.get('lat') if request.is_json else None 

106 lng = request.form.get('lng') or request.json.get('lng') if request.is_json else None 

107 item_type = (request.form.get('item_type') or (request.json.get('item_type') if request.is_json else '')).strip() 

108 item_name = (request.form.get('item_name') or (request.json.get('item_name') if request.is_json else '')).strip() 

109 if not item_name: 

110 item_name = item_type 

111 if not lat or not lng or not item_type: 

112 return jsonify({'success': False, 'message': 'Nedostaju podaci.'}), 400 

113 

114 # Use secure coordinate validation 

115 valid, error_msg, coordinates = validate_geographic_coordinates(lat, lng) 

116 if not valid: 

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

118 lat_f, lng_f = coordinates 

119 db = get_db() 

120 _ensure_map_schema(db) 

121 try: 

122 cur = db.execute('INSERT INTO markers (user_id, type, latitude, longitude, name) VALUES (?, ?, ?, ?, ?)', (session['user_id'], item_type, lat_f, lng_f, item_name)) 

123 db.commit() 

124 return jsonify({'success': True, 'item_name': item_name, 'item_id': cur.lastrowid}) 

125 except sqlite3.Error: 

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

127 

128 

129@bp.route('/get_marker_locations') 

130@login_required 

131def get_marker_locations(): 

132 db = get_db() 

133 _ensure_map_schema(db) 

134 result = [] 

135 # Generic markers 

136 for m in db.execute('SELECT id, type, latitude, longitude, name FROM markers WHERE user_id=?', (session['user_id'],)).fetchall(): 

137 result.append({ 

138 'id': m['id'], 

139 'type': m['type'], 

140 'latitude': m['latitude'], 

141 'longitude': m['longitude'], 

142 'name': m['name'] or m['type'] 

143 }) 

144 # Camera markers stored in cameras table 

145 for c in db.execute('SELECT camera_id, camera_name, location FROM cameras WHERE user_id=? AND IFNULL(added_to_map,0)=1 AND IFNULL(location,"")<>""', (session['user_id'],)).fetchall(): 

146 try: 

147 lat_s, lng_s = (c['location'] or '').split(',') 

148 lat_f = float(lat_s); lng_f = float(lng_s) 

149 except Exception: 

150 continue 

151 result.append({ 

152 'id': str(c['camera_id']), 

153 'type': 'camera', 

154 'latitude': lat_f, 

155 'longitude': lng_f, 

156 'name': c['camera_name'] 

157 }) 

158 return jsonify(result) 

159 

160 

161@bp.route('/delete_marker', methods=['POST']) 

162@login_required 

163def delete_marker(): 

164 item_id = (request.form.get('item_id') or (request.json.get('item_id') if request.is_json else '')).strip() 

165 item_type = (request.form.get('item_type') or (request.json.get('item_type') if request.is_json else '')).strip() 

166 if not item_id: 

167 return jsonify({'success': False, 'message': 'Nedostaju podaci.'}), 400 

168 db = get_db() 

169 _ensure_map_schema(db) 

170 try: 

171 if item_type == 'camera': 

172 db.execute('UPDATE cameras SET location=NULL, added_to_map=0 WHERE user_id=? AND camera_id=?', (session['user_id'], item_id)) 

173 db.commit() 

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

175 else: 

176 db.execute('DELETE FROM markers WHERE id=? AND user_id=?', (item_id, session['user_id'])) 

177 db.commit() 

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

179 except sqlite3.Error: 

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

181 

182