c7b2f5990f5e416cfee933da916bc50a673cbf60
[mediacube.git] /
1 package user.jobengine.server.steps;\r
2 \r
3 import java.io.OutputStream;\r
4 import java.net.InetAddress;\r
5 import java.net.URI;\r
6 import java.text.ParseException;\r
7 import java.text.SimpleDateFormat;\r
8 import java.util.Calendar;\r
9 import java.util.Date;\r
10 import java.util.List;\r
11 import java.util.TimeZone;\r
12 \r
13 import org.apache.commons.lang.StringUtils;\r
14 import org.apache.commons.net.ftp.FTP;\r
15 import org.apache.commons.net.ftp.FTPClient;\r
16 import org.apache.commons.net.ftp.FTPReply;\r
17 import org.apache.logging.log4j.LogManager;\r
18 import org.apache.logging.log4j.Logger;\r
19 import org.apache.logging.log4j.Marker;\r
20 \r
21 import com.ibm.nosql.json.api.BasicDBList;\r
22 import com.ibm.nosql.json.api.BasicDBObject;\r
23 import com.ibm.nosql.json.api.DB;\r
24 import com.ibm.nosql.json.api.DBCollection;\r
25 import com.ibm.nosql.json.api.DBCursor;\r
26 import com.ibm.nosql.json.api.DBObject;\r
27 import com.ibm.nosql.json.api.QueryBuilder;\r
28 \r
29 import user.commons.CalendarUtils;\r
30 import user.commons.ListUtils;\r
31 import user.commons.StoreUri;\r
32 import user.commons.configuration.SystemConfiguration;\r
33 import user.commons.nexio.NexioDispatcher;\r
34 import user.commons.nosql.NoSQLUtils;\r
35 import user.commons.octopus.IOctopusAPI;\r
36 import user.commons.octopus.OctopusAPI;\r
37 import user.commons.remotestore.FtpDirectoryLister;\r
38 import user.commons.remotestore.RemoteStoreProtocol;\r
39 import user.jobengine.db.IItemManager;\r
40 import user.jobengine.server.IJobEngine;\r
41 import user.jobengine.server.IJobRuntime;\r
42 import user.jobengine.server.steps.shared.EscortFiles;\r
43 \r
44 public class CopyForArchiveNEXIORecordingsStep extends JobStep {\r
45         private static final String MEDIATYPE = "Visszarögzített";\r
46         private static final String SCHEDULED_FORMAT = "yyyy.MM.dd HH:mm";\r
47         private static final String STARTTIME_FORMAT = "HHmm";\r
48         private static final String RUNDOWNDATE_FORMAT = "yyyyMMdd";\r
49         private static final Logger logger = LogManager.getLogger();\r
50         private static final String UTF_8 = "utf-8";\r
51         private static final String JSON_EXT = ".json";\r
52         private static final String XML_EXT = ".xml";\r
53         private static final String DURATION = "duration";\r
54         private static final String MXFEXT = ".MXF";\r
55         private static final String NEXIOCLIPS = NexioDispatcher.CLIP_COLLECTION_NAME;\r
56         private static final String LONGNAMEID = "longnameid";\r
57         private static final String EXTAGENCY = "extagency";\r
58         private static final String RECORDDATE = "recorddate";\r
59         private static final SimpleDateFormat startTimeformat = new SimpleDateFormat(STARTTIME_FORMAT);\r
60         private static final SimpleDateFormat rundownDateformat = new SimpleDateFormat(RUNDOWNDATE_FORMAT);\r
61         private static String NEXIO_HOST = SystemConfiguration.getInstance().value("services.nexio.host");\r
62 \r
63         private OctopusAPI octopusAPI;\r
64         private IItemManager manager;\r
65 \r
66         private DB db;\r
67         private FTPClient sourceFtp;\r
68         private FTPClient targetFtp;\r
69         private StoreUri sourceUri;\r
70         private StoreUri targetUri;\r
71         private int nexioKillDateDays;\r
72         private String nexioAgency;\r
73         private Object[] filterAgencies;\r
74         private Marker systemMarker;\r
75 \r
76         private int check(int value, String name) {\r
77                 if (value == 0) {\r
78                         logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere 0.", name);\r
79                         throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));\r
80                 }\r
81                 return value;\r
82         }\r
83 \r
84         private String check(String value, String name) {\r
85                 if (StringUtils.isBlank(value)) {\r
86                         logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere üres.", name);\r
87                         throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));\r
88                 }\r
89                 return value;\r
90         }\r
91 \r
92         private void copy(RundownArchive rundownArchive) {\r
93                 for (StoryArchive storyArchive : rundownArchive.getStoryArchives()) {\r
94                         for (FileArchive fileArchive : storyArchive.getFileArchives()) {\r
95                                 try {\r
96                                         copyFile(fileArchive, rundownArchive, storyArchive);\r
97                                         logger.info(systemMarker, "Sikeres fájl archiválás a '{}' tükörhöz: '{}'.", rundownArchive.getItemTitle(), fileArchive.getFileName());\r
98                                 } catch (Exception e) {\r
99                                         logger.error(systemMarker, "Az '{}' fájl archiválása sikertelen. A rendszer üzenete: {}", fileArchive.getFileName(), e.getMessage());\r
100                                 }\r
101                         }\r
102                 }\r
103         }\r
104 \r
105         private void copyFile(FileArchive fileArchive, RundownArchive rundownArchive, StoryArchive storyArchive) throws Exception {\r
106                 String sourceFileName = fileArchive.getFileName();\r
107                 String targetFileName = getTargetFileName(rundownArchive, sourceFileName);\r
108                 transferFile(sourceFileName, targetFileName);\r
109                 BasicDBObject metadata = createMetadata(rundownArchive, storyArchive, fileArchive);\r
110                 transferMetadata(targetFileName, metadata);\r
111                 createSourceKillDateFile(rundownArchive, sourceFileName);\r
112         }\r
113 \r
114         private BasicDBObject createMetadata(RundownArchive rundownArchive, StoryArchive storyArchive, FileArchive fileArchive) {\r
115                 BasicDBObject result = new BasicDBObject();\r
116                 result.put("itemHouseId", rundownArchive.getItemHouseId());\r
117                 result.put("itemTitle", rundownArchive.getItemTitle());\r
118                 result.put("itemDescription", rundownArchive.getItemDesc());\r
119                 result.put("userName", "mediacube");\r
120 \r
121                 result.put("mediaHouseId", storyArchive.getMediaHouseId());\r
122                 result.put("mediaTitle", storyArchive.getMediaTitle());\r
123                 result.put("mediaDescription", storyArchive.getMediaDesc());\r
124 \r
125                 result.put("duration", fileArchive.getDuration());\r
126                 result.put("mediaType", MEDIATYPE);\r
127                 return result;\r
128         }\r
129 \r
130         private void createSourceKillDateFile(RundownArchive rundownArchive, String fileName) throws Exception {\r
131                 logger.info("Create killdate/agency for {}", fileName);\r
132                 OutputStream outStream = null;\r
133                 try {\r
134                         sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();\r
135                         Calendar killDate = CalendarUtils.createCalendar(rundownArchive.getScheduleDate());\r
136                         killDate.add(Calendar.DAY_OF_YEAR, nexioKillDateDays);\r
137                         byte[] killDateFile = EscortFiles.createNEXIOKillDateFile(fileName, killDate.getTime(), null, nexioAgency);\r
138                         outStream = sourceFtp.storeFileStream(fileName + XML_EXT);\r
139                         if (outStream == null) {\r
140                                 throw new NullPointerException("Can not open: " + fileName + XML_EXT + " Reply:" + sourceFtp.getReplyString());\r
141                         }\r
142                         outStream.write(killDateFile);\r
143                         outStream.flush();\r
144                 } catch (Exception e) {\r
145                         logger.catching(e);\r
146                         throw e;\r
147                 } finally {\r
148                         if (outStream != null)\r
149                                 outStream.close();\r
150                         sourceUri.cleanUp();\r
151                 }\r
152         }\r
153 \r
154         @StepEntry\r
155         public Object[] execute(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,\r
156                         String filterAgencies, int limit, int nexioKillDateDays, String targetAgency, IJobEngine jobEngine, IJobRuntime jobRuntime) throws Exception {\r
157                 systemMarker = jobRuntime.getSessionMarker();\r
158                 setAndCheck(nexioPort, nexioUserName, nexioPassword, archiveFtp, archiveUserName, archivePassword, filterAgencies, limit, nexioKillDateDays,\r
159                                 targetAgency, jobEngine);\r
160                 octopusAPI = new OctopusAPI();\r
161                 List<BasicDBObject> clips = queryClips();\r
162                 processClips(clips, limit);\r
163                 return null;\r
164         }\r
165 \r
166         private Date getScheduledStart(String clipName, Date recordDate) {\r
167                 if (StringUtils.isBlank(clipName)) {\r
168                         logger.warn(systemMarker, "A fájlnak nincs neve, ezért nem archiválható.");\r
169                         return null;\r
170                 }\r
171                 if (recordDate == null) {\r
172                         logger.warn(systemMarker, "Az '{}' fájl rögzítésének ideje nem meghatározható, ezért nem archiválható.", clipName);\r
173                         return null;\r
174                 }\r
175 \r
176                 Date timePart = null;\r
177                 try {\r
178                         String clipNameTime = clipName.split("_")[0];\r
179                         timePart = startTimeformat.parse(clipNameTime);\r
180                 } catch (ParseException e) {\r
181                         logger.warn(systemMarker, "A '{}' fájl neve nem időbélyeggel kezdődik, ezért nem archiválható.", clipName);\r
182                         return null;\r
183                 }\r
184                 Calendar dateCal = CalendarUtils.createCalendar(recordDate);\r
185                 dateCal.setTimeZone(TimeZone.getTimeZone("Europe/Budapest"));\r
186                 Calendar wholeCal = CalendarUtils.createCalendar(CalendarUtils.createCalendar(recordDate), timePart);\r
187                 wholeCal.setTimeZone(TimeZone.getTimeZone("Europe/Budapest"));\r
188                 return wholeCal.getTime();\r
189         }\r
190 \r
191         private String getTargetFileName(RundownArchive rundownArchive, String sourceFileName) {\r
192                 String date = rundownDateformat.format(rundownArchive.getScheduleDate());\r
193                 return String.format("%s-%s%s", date, sourceFileName, MXFEXT);\r
194         }\r
195 \r
196         private RundownArchive processClip(BasicDBObject clip) {\r
197                 RundownArchive result = null;\r
198                 String clipName = NoSQLUtils.asString(clip, LONGNAMEID);\r
199                 Date recordDate = clip.getDate(RECORDDATE);\r
200                 long duration = NoSQLUtils.asLong(clip, DURATION);\r
201                 logger.info("Processing {} {}", clipName, recordDate);\r
202                 Date scheduledStart = getScheduledStart(clipName, recordDate);\r
203                 if (scheduledStart == null)\r
204                         return null;\r
205                 DBObject rundown = octopusAPI.getRundown(scheduledStart);\r
206                 if (rundown == null) {\r
207                         logger.error(systemMarker, "A '{}' anyaghoz nem található tükör '{}' kezdéssel, ezért nem archiválható.", clipName, scheduledStart);\r
208                         return null;\r
209                 }\r
210 \r
211                 try {\r
212                         result = processRundow(rundown, clipName, duration);\r
213                 } catch (Exception e) {\r
214                         logger.catching(e);\r
215                         logger.error(systemMarker, "A '{}' anyag metaadatainak transzformálása sikertelen, ezért az nem archiválható. A rendszer hibaüzenete: {}",\r
216                                         e.getMessage());\r
217                         return null;\r
218                 }\r
219                 /*\r
220                                 if (clipName.startsWith("1900_")) {\r
221                                         String clipNameNext = clipName.replace("1900_", "1908_");\r
222                                         scheduledStart = getScheduledStart(clipNameNext, recordDate);\r
223                                         rundown = octopusAPI.getRundown(scheduledStart);\r
224                                         if (rundown == null) {\r
225                                                 Calendar calendar = CalendarUtils.createCalendar(scheduledStart);\r
226                                                 int dow = calendar.get(Calendar.DAY_OF_WEEK);\r
227                                                 if (dow == Calendar.SATURDAY || dow == Calendar.SUNDAY) {\r
228                                                         logger.info(systemMarker, "A '{}' anyaghoz nem található tükör '{}' kezdéssel, de a hétvégi kivétel miatt archiválható.", clipName,\r
229                                                                         scheduledStart);\r
230                                                         return result;\r
231                                                 } else {\r
232                                                         logger.error(systemMarker, "A '{}' anyaghoz nem található tükör '{}' kezdéssel, ezért nem archiválható.", clipName, scheduledStart);\r
233                                                         return null;\r
234                                                 }\r
235                                         }\r
236                 \r
237                                         RundownArchive item2 = null;\r
238                 \r
239                                         try {\r
240                                                 item2 = processRundow(rundown, clipName, duration);\r
241                                         } catch (Exception e) {\r
242                                                 logger.catching(e);\r
243                                                 logger.error(systemMarker, "A '{}' anyag metaadatainak transzformálása sikertelen, ezért az nem archiválható. A rendszer hibaüzenete: {}",\r
244                                                                 e.getMessage());\r
245                                                 return null;\r
246                                         }\r
247                 \r
248                                         result.setItemTitle(result.getItemTitle() + " + NAPIAKT");\r
249                                         StoryArchive storyArchive = result.getStoryArchives().get(0);\r
250                                         StoryArchive storyArchive2 = item2.getStoryArchives().get(0);\r
251                                         storyArchive.setMediaDesc(storyArchive.getMediaDesc() + "\r\n\r\n****** NAPIAKT ******\r\n\r\n" + storyArchive2.getMediaDesc());\r
252                                 }\r
253                 */\r
254                 return result;\r
255         }\r
256 \r
257         private void processClips(List<BasicDBObject> clips, int limit) {\r
258                 logger.info(systemMarker, "A folyamat {} archiválható anyagot érzékelt.", clips.size());\r
259                 int current = 0;\r
260                 for (BasicDBObject clip : clips) {\r
261                         RundownArchive rundownArchive = processClip(clip);\r
262                         if (rundownArchive == null)\r
263                                 continue;\r
264 \r
265                         copy(rundownArchive);\r
266 \r
267                         current++;\r
268                         if (current == limit)\r
269                                 break;\r
270                         setProgress(current * 100 / limit);\r
271                 }\r
272         }\r
273 \r
274         private RundownArchive processRundow(DBObject r, String clipName, long duration) throws Exception {\r
275                 BasicDBObject rundown = (BasicDBObject) r;\r
276                 long rundownID = rundown.getLong(IOctopusAPI.ID);\r
277                 logger.info("Processing rundown {} {}", rundownID, rundown.getString(IOctopusAPI.NAME));\r
278 \r
279                 List<DBObject> stories = octopusAPI.getRundownFullStories(rundownID);\r
280                 if (stories == null)\r
281                         return null;\r
282                 RundownArchive result = new RundownArchive();\r
283 \r
284                 String name = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.RUNDOWN_TYPE), IOctopusAPI.NAME);\r
285                 if (StringUtils.isBlank(name))\r
286                         return null;\r
287                 String channel = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.CHANNEL), IOctopusAPI.NAME);\r
288                 Date scheduledStart = rundown.getDate(IOctopusAPI.SCHEDULED_START);\r
289                 if (scheduledStart == null)\r
290                         return null;\r
291                 String start = CalendarUtils.toString(CalendarUtils.createCalendar(scheduledStart), SCHEDULED_FORMAT);\r
292                 result.setScheduleDate(scheduledStart);\r
293                 result.setItemHouseId(String.valueOf(rundownID));\r
294                 result.setItemTitle(String.format("%s %s %s", start, name, channel));\r
295 \r
296                 StoryArchive storyArchive = new StoryArchive();\r
297                 storyArchive.setMediaHouseId(result.getItemHouseId());\r
298                 storyArchive.setMediaTitle(clipName);\r
299                 storyArchive.setMediaDesc(octopusAPI.getRundownContent(stories));\r
300                 result.addStoryArchive(storyArchive);\r
301                 storyArchive.addFileArchive(new FileArchive(clipName, duration));\r
302                 return result;\r
303         }\r
304 \r
305         private List<BasicDBObject> queryClips() {\r
306                 DBCollection collection = db.getCollection(NEXIOCLIPS);\r
307                 BasicDBList agencies = new BasicDBList(filterAgencies);\r
308 \r
309                 Calendar now = CalendarUtils.createZeroCalendar();\r
310                 QueryBuilder dateQueryBuilder = QueryBuilder.start(RECORDDATE).lessThan(now.getTime());\r
311                 QueryBuilder queryBuilder = QueryBuilder.start(EXTAGENCY).in(agencies).and(dateQueryBuilder.get());\r
312                 DBCursor cursor = collection.find(queryBuilder.get()).sort(RECORDDATE, -1);\r
313                 if (!cursor.hasNext())\r
314                         return null;\r
315                 return ListUtils.cast(cursor.toArray());\r
316         }\r
317 \r
318         private void setAndCheck(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,\r
319                         String agencies, int limit, int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine) throws Exception {\r
320                 db = NoSQLUtils.getNoSQLDB();\r
321                 if (db == null) {\r
322                         logger.error(systemMarker, "Az NoSQL adatkezelő réteg nem elérhető.");\r
323                         throw new NullPointerException("Internal error, missing NoSQL DB reference.");\r
324                 }\r
325 \r
326                 if (jobEngine == null) {\r
327                         logger.error(systemMarker, "Az folyamatkezelő réteg nem elérhető.");\r
328                         throw new NullPointerException("Internal error, missing JobEngine reference.");\r
329                 }\r
330                 manager = jobEngine.getItemManager();\r
331                 if (manager == null) {\r
332                         logger.error(systemMarker, "Az adatbáziskezelő réteg nem elérhető.");\r
333                         throw new NullPointerException("Internal error, missing ItemManager reference.");\r
334                 }\r
335 \r
336                 if (StringUtils.isBlank(NEXIO_HOST)) {\r
337                         logger.error(systemMarker, "A 'nexio.host' rendszer paraméter nem található.");\r
338                         throw new NullPointerException("System is not configured properly, 'jobengine.selenio.address' startup parameter missing.");\r
339                 }\r
340                 check(nexioPort, "nexioPort");\r
341                 check(nexioUserName, "nexioUserName");\r
342                 check(nexioPassword, "nexioPassword");\r
343 \r
344                 check(agencies, "filterAgencies");\r
345                 filterAgencies = agencies.split(",");\r
346 \r
347                 check(limit, "limit");\r
348 \r
349                 check(nexioKillDateDays, "nexioKillDateDays");\r
350                 this.nexioKillDateDays = nexioKillDateDays;\r
351                 check(nexioAgency, "nexioAgency");\r
352                 this.nexioAgency = nexioAgency;\r
353 \r
354                 sourceUri = manager.createStoreUri(RemoteStoreProtocol.FTP, NEXIO_HOST);\r
355                 sourceUri.setPortNumber(nexioPort);\r
356                 sourceUri.setUserName(nexioUserName);\r
357                 sourceUri.setPassword(nexioPassword);\r
358                 if (sourceUri == null) {\r
359                         logger.error(systemMarker, "A forrás nem elérhető.");\r
360                         throw new NullPointerException("Internal error, missing 'sourceUri'.");\r
361                 }\r
362 \r
363                 check(archiveFtp, "archiveFtp");\r
364                 check(archiveUserName, "archiveUserName");\r
365                 check(archivePassword, "archivePassword");\r
366 \r
367                 targetUri = manager.createStoreUri(new URI(archiveFtp));\r
368                 targetUri.setUserName(archiveUserName);\r
369                 targetUri.setPassword(archivePassword);\r
370                 if (targetUri == null) {\r
371                         logger.error(systemMarker, "A cél nem elérhető.");\r
372                         throw new NullPointerException("Internal error, missing 'targetUri'.");\r
373                 }\r
374 \r
375         }\r
376 \r
377         private void transferFile(String sourceFileName, String targetFileName) throws Exception {\r
378                 int reply = 0;\r
379                 logger.info("Transfer clip {}", sourceFileName);\r
380                 try {\r
381                         sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();\r
382                         targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
383 \r
384                         if (!targetFtp.enterRemotePassiveMode())\r
385                                 throw new Exception("!PASV");\r
386 \r
387                         reply = sourceFtp.port(InetAddress.getByName(targetFtp.getPassiveHost()), targetFtp.getPassivePort());\r
388                         if (!FTPReply.isPositiveCompletion(reply))\r
389                                 throw new Exception("PORT parancs válasza: " + sourceFtp.getReplyString());\r
390 \r
391                         if (!sourceFtp.setFileType(FTP.BINARY_FILE_TYPE))\r
392                                 throw new Exception("!SOURCE TYPE");\r
393 \r
394                         reply = sourceFtp.retr(sourceFileName);\r
395                         if (!FTPReply.isPositivePreliminary(reply))\r
396                                 throw new Exception("RETR parancs válasza: " + sourceFtp.getReplyString());\r
397 \r
398                         if (!targetFtp.setFileType(FTP.BINARY_FILE_TYPE))\r
399                                 throw new Exception("!TARGET TYPE");\r
400 \r
401                         reply = targetFtp.stor(targetFileName);\r
402                         if (!FTPReply.isPositivePreliminary(reply))\r
403                                 throw new Exception("STOR parancs válasza: " + sourceFtp.getReplyString());\r
404 \r
405                         while (true) {\r
406                                 reply = sourceFtp.stat();\r
407                                 if (!FTPReply.isPositiveCompletion(reply))\r
408                                         throw new Exception("STAT parancs válasza: " + sourceFtp.getReplyString());\r
409 \r
410                                 //logger.info("Status: {}", sourceFtp.getReplyString());\r
411                                 if (reply == 226) {\r
412                                         break;\r
413                                 }\r
414                                 Thread.sleep(1000);\r
415                         }\r
416                 } catch (Exception e) {\r
417                         logger.catching(e);\r
418                         throw e;\r
419                 } finally {\r
420                         sourceUri.cleanUp();\r
421                         targetUri.cleanUp();\r
422                 }\r
423 \r
424         }\r
425 \r
426         private void transferMetadata(String fileName, BasicDBObject metadata) throws Exception {\r
427                 logger.info("Transfer metadata {}", fileName);\r
428                 OutputStream outStream = null;\r
429                 try {\r
430                         targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
431                         if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER)) {\r
432                                 targetFtp.makeDirectory(EscortFiles.STATUSFOLDER);\r
433                                 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER))\r
434                                         throw new Exception("!STATUSFOLDER");\r
435                         }\r
436 \r
437                         outStream = targetFtp.storeFileStream(fileName + JSON_EXT);\r
438                         if (outStream == null) {\r
439                                 throw new NullPointerException("Can not open: " + fileName + JSON_EXT + " Reply:" + targetFtp.getReplyString());\r
440                         }\r
441                         outStream.write(metadata.toString().getBytes(UTF_8));\r
442                         outStream.flush();\r
443                         //targetFtp.changeToParentDirectory();\r
444                 } catch (Exception e) {\r
445                         logger.catching(e);\r
446                         throw e;\r
447                 } finally {\r
448                         if (outStream != null)\r
449                                 outStream.close();\r
450                         targetUri.cleanUp();\r
451                 }\r
452         }\r
453 \r
454 }\r