b69535f37a9d43cb6335237af4b27e2a69be8541
[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.util.Calendar;\r
7 import java.util.Date;\r
8 import java.util.List;\r
9 \r
10 import org.apache.commons.lang.StringUtils;\r
11 import org.apache.commons.net.ftp.FTP;\r
12 import org.apache.commons.net.ftp.FTPClient;\r
13 import org.apache.commons.net.ftp.FTPReply;\r
14 import org.apache.logging.log4j.LogManager;\r
15 import org.apache.logging.log4j.Logger;\r
16 import org.apache.logging.log4j.Marker;\r
17 \r
18 import com.ibm.nosql.json.api.BasicDBObject;\r
19 import com.ibm.nosql.json.api.DB;\r
20 import com.ibm.nosql.json.api.DBCollection;\r
21 import com.ibm.nosql.json.api.DBCursor;\r
22 import com.ibm.nosql.json.api.DBObject;\r
23 \r
24 import user.commons.CalendarUtils;\r
25 import user.commons.ListUtils;\r
26 import user.commons.StoreUri;\r
27 import user.commons.nosql.NoSQLUtils;\r
28 import user.commons.octopus.IOctopusAPI;\r
29 import user.commons.octopus.OctopusAPI;\r
30 import user.commons.remotestore.FtpDirectoryLister;\r
31 import user.commons.remotestore.RemoteStoreProtocol;\r
32 import user.jobengine.db.IItemManager;\r
33 import user.jobengine.server.IJobEngine;\r
34 import user.jobengine.server.IJobRuntime;\r
35 import user.jobengine.server.steps.MetadataTypeDetector.MetadataType;\r
36 \r
37 public class CopyForArchiveNEXIOMaterialsStep extends JobStep {\r
38         private static final String SCHEDULED_FORMAT = "yyyy.MM.dd HH:mm";\r
39         private static final Logger logger = LogManager.getLogger();\r
40         private static final String UTF_8 = "utf-8";\r
41         private static final String JSON_EXT = ".json";\r
42         private static final String XML_EXT = ".xml";\r
43         private static final String DURATION = "duration";\r
44         private static final String MXFEXT = ".MXF";\r
45         private static final String NEXIOCLIPS = "nexioclips";\r
46         private static final String LONGNAMEID = "longnameid";\r
47         private static final String ARCHIVEDRUNDOWNS = "archivedrundowns";\r
48         private static final String ID = "id";\r
49         private static final String MEDIATYPE = "Hír bejátszó";\r
50         private OctopusAPI octopusAPI;\r
51         private IItemManager manager;\r
52 \r
53         private DB db;\r
54         private FTPClient sourceFtp;\r
55         private FTPClient targetFtp;\r
56         private StoreUri sourceUri;\r
57         private StoreUri targetUri;\r
58         private int nexioKillDateDays;\r
59         private String nexioAgency;\r
60         private Marker systemMarker;\r
61 \r
62         private int check(int value, String name) {\r
63                 if (value == 0) {\r
64                         logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere 0.", name);\r
65                         throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));\r
66                 }\r
67                 return value;\r
68         }\r
69 \r
70         private String check(String value, String name) {\r
71                 if (value == null) {\r
72                         logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere üres.", name);\r
73                         throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));\r
74                 }\r
75                 return value;\r
76         }\r
77 \r
78         private void copy(RundownArchive rundownArchive) throws Exception {\r
79                 for (StoryArchive storyArchive : rundownArchive.getStoryArchives()) {\r
80                         for (FileArchive fileArchive : storyArchive.getFileArchives()) {\r
81                                 try {\r
82                                         copyFile(fileArchive, rundownArchive, storyArchive);\r
83                                 } catch (Exception e) {\r
84                                         logger.error(systemMarker, "A '{}' clip archiválása sikertelen. A rendszer üzenete: {}", fileArchive.getFileName(), e.getMessage());\r
85                                 }\r
86                         }\r
87                 }\r
88         }\r
89 \r
90         private void copyFile(FileArchive fileArchive, RundownArchive rundownArchive, StoryArchive storyArchive) throws Exception {\r
91                 String origFileName = fileArchive.getFileName();\r
92                 String fileName = String.format("%s-%s", origFileName, rundownArchive.getItemHouseId());\r
93                 String videoFileName = fileName + MXFEXT;\r
94 \r
95                 //A mar letezo mozikat nem archivaljuk le ujra, csak a metaadatot\r
96                 long existingMediaId = ItemManagerExtensions.getExistingRundownMedia(manager, origFileName);\r
97                 if (existingMediaId == 0)\r
98                         transferFile(origFileName + MXFEXT, videoFileName);\r
99                 else\r
100                         transferChunk(videoFileName);\r
101 \r
102                 BasicDBObject metadata = createMetadata(rundownArchive, storyArchive, fileArchive, existingMediaId);\r
103                 transferMetadata(videoFileName, metadata);\r
104                 createSourceKillDateFile(rundownArchive, origFileName);\r
105         }\r
106 \r
107         private BasicDBObject createMetadata(RundownArchive rundownArchive, StoryArchive storyArchive, FileArchive fileArchive, long existingMediaId) {\r
108                 BasicDBObject result = new BasicDBObject();\r
109                 result.put("itemHouseId", rundownArchive.getItemHouseId());\r
110                 result.put("itemTitle", rundownArchive.getItemTitle());\r
111                 result.put("itemDescription", rundownArchive.getItemDesc());\r
112                 result.put("userName", "mediacube");\r
113                 result.put("mediaHouseId", storyArchive.getMediaHouseId());\r
114                 result.put("mediaTitle", storyArchive.getMediaTitle());\r
115                 result.put("mediaDescription", storyArchive.getMediaDesc());\r
116                 result.put("mediaType", MEDIATYPE);\r
117                 result.put("duration", fileArchive.getDuration());\r
118                 result.put("existingMediaId", existingMediaId);\r
119                 return result;\r
120         }\r
121 \r
122         private void createSourceKillDateFile(RundownArchive rundownArchive, String fileName) throws Exception {\r
123                 logger.info("Create killdate/agency for {}", fileName);\r
124                 OutputStream outStream = null;\r
125                 try {\r
126                         sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();\r
127                         Calendar killDate = CalendarUtils.createCalendar(rundownArchive.getScheduleDate());\r
128                         killDate.add(Calendar.DAY_OF_YEAR, nexioKillDateDays);\r
129                         byte[] killDateFile = EscortFiles.createNEXIOKillDateFile(fileName, killDate.getTime(), null, nexioAgency);\r
130                         outStream = sourceFtp.storeFileStream(fileName + XML_EXT);\r
131                         if (outStream == null) {\r
132                                 throw new NullPointerException("Can not open: " + fileName + XML_EXT + " Reply:" + sourceFtp.getReplyString());\r
133                         }\r
134                         outStream.write(killDateFile);\r
135                         outStream.flush();\r
136                 } catch (Exception e) {\r
137                         logger.catching(e);\r
138                         throw e;\r
139                 } finally {\r
140                         if (outStream != null)\r
141                                 outStream.close();\r
142                         sourceUri.cleanUp();\r
143                 }\r
144         }\r
145 \r
146         @StepEntry\r
147         public Object[] execute(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,\r
148                         int daysBeforeNow, int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine, IJobRuntime jobRuntime) throws Exception {\r
149                 systemMarker = jobRuntime.getMarker();\r
150                 setAndCheck(nexioPort, nexioUserName, nexioPassword, archiveFtp, archiveUserName, archivePassword, nexioKillDateDays, nexioAgency, jobEngine);\r
151                 octopusAPI = new OctopusAPI();\r
152                 Calendar scheduledDate = Calendar.getInstance();\r
153                 scheduledDate.add(Calendar.DAY_OF_YEAR, -1 * daysBeforeNow);\r
154                 List<DBObject> rundowns = octopusAPI.getRundowns(scheduledDate.getTime());\r
155                 if (rundowns == null) {\r
156                         logger.warn(systemMarker, "Nem található adástükör a {} napra.", CalendarUtils.toDateString(scheduledDate));\r
157                         return null;\r
158                 }\r
159 \r
160                 processRundowns(rundowns);\r
161                 return null;\r
162         }\r
163 \r
164         //      private String getVersionedFileName(String fileName, String extension) throws Exception {\r
165         //              String result = fileName;\r
166         //              try {\r
167         //                      targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
168         //                      FTPFile[] listFiles = targetFtp.listFiles();\r
169         //                      List<String> fileNames = new ArrayList<>();\r
170         //                      for (FTPFile ftpFile : listFiles) {\r
171         //                              fileNames.add(ftpFile.getName());\r
172         //                      }\r
173         //                      while (fileNames.contains(result + extension)) {\r
174         //\r
175         //                      }\r
176         //\r
177         //              } catch (Exception e) {\r
178         //                      logger.catching(e);\r
179         //                      throw e;\r
180         //              } finally {\r
181         //                      targetUri.cleanUp();\r
182         //              }\r
183         //              return result;\r
184         //      }\r
185 \r
186         private FileArchive processMosObject(BasicDBObject rundown, BasicDBObject story, BasicDBObject mosObject) throws Exception {\r
187                 String mosID = mosObject.getString(IOctopusAPI.OBJ_ID);\r
188                 if (MetadataTypeDetector.GuessMetadataType(mosID) != MetadataType.OctopusPlaceholder) {\r
189                         logger.trace("Skipping MOS object {}", mosID);\r
190                         return null;\r
191                 }\r
192                 DBCollection clips = db.getCollection(NEXIOCLIPS);\r
193                 BasicDBObject clip = (BasicDBObject) clips.findOne(new BasicDBObject(LONGNAMEID, mosID));\r
194                 if (clip == null) {\r
195                         logger.debug("File {} NOT exists", mosID);\r
196                         return null;\r
197                         //throw new Exception(String.format("File NOT exists %s", mosID));\r
198                 } else {\r
199                         logger.debug("File {} exists", mosID);\r
200                 }\r
201                 long duration = NoSQLUtils.asLong(clip, DURATION);\r
202 \r
203                 if (duration == 0) {\r
204                         logger.debug("File {} exists with 0 frame length", mosID);\r
205                         return null;\r
206                 }\r
207                 if (duration == 1) {\r
208                         logger.debug("File {} exists with 1 frame length", mosID);\r
209                         return null;\r
210                 }\r
211                 return new FileArchive(mosID, duration);\r
212         }\r
213 \r
214         private RundownArchive processRundow(DBObject r) throws Exception {\r
215                 BasicDBObject rundown = (BasicDBObject) r;\r
216                 long rundownID = rundown.getLong(ID);\r
217                 logger.info("Processing rundown {} {}", rundownID, rundown.getString(IOctopusAPI.NAME));\r
218                 List<DBObject> stories = octopusAPI.getRundownFullStories(rundownID);\r
219                 if (stories == null)\r
220                         return null;\r
221                 RundownArchive result = new RundownArchive();\r
222 \r
223                 long id = NoSQLUtils.asLong(rundown, IOctopusAPI.ID);\r
224                 if (id == 0)\r
225                         return null;\r
226                 String name = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.RUNDOWN_TYPE), IOctopusAPI.NAME);\r
227                 if (StringUtils.isBlank(name))\r
228                         return null;\r
229                 String channel = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.CHANNEL), IOctopusAPI.NAME);\r
230                 Date scheduledStart = rundown.getDate(IOctopusAPI.SCHEDULED_START);\r
231                 if (scheduledStart == null)\r
232                         return null;\r
233                 String start = CalendarUtils.toString(CalendarUtils.createCalendar(scheduledStart), SCHEDULED_FORMAT);\r
234                 result.setScheduleDate(scheduledStart);\r
235                 result.setItemHouseId(String.valueOf(id));\r
236                 result.setItemTitle(String.format("%s %s %s", start, name, channel));\r
237 \r
238                 for (DBObject s : stories) {\r
239                         StoryArchive storyArchive = processStory(rundown, s);\r
240                         if (storyArchive == null)\r
241                                 continue;\r
242                         result.addStoryArchive(storyArchive);\r
243                 }\r
244                 return result;\r
245         }\r
246 \r
247         private void processRundowns(List<DBObject> rundowns) {\r
248 \r
249                 //TODO kiveni publikálás előtt\r
250                 //db.getCollection(ARCHIVEDRUNDOWNS).drop();\r
251                 List<BasicDBObject> archivedRundowns = queryArchivedRundowns();\r
252 \r
253                 int index = 1;\r
254 \r
255                 for (DBObject r : rundowns) {\r
256                         BasicDBObject rundown = (BasicDBObject) r;\r
257                         setProgress(index * 100 / rundowns.size());\r
258                         long rundownID = NoSQLUtils.asLong(rundown, IOctopusAPI.ID);\r
259                         String rundownName = NoSQLUtils.asString(rundown, IOctopusAPI.NAME);\r
260                         try {\r
261                                 BasicDBObject currentRundownID = new BasicDBObject(IOctopusAPI.ID, rundownID);\r
262                                 if (archivedRundowns != null && archivedRundowns.contains(currentRundownID)) {\r
263                                         logger.info("Skipping archived rundown {} {}", rundownID, rundownName);\r
264                                         continue;\r
265                                 }\r
266 \r
267                                 RundownArchive rundownArchive = processRundow(r);\r
268                                 if (rundownArchive == null || rundownArchive.isEmpty()) {\r
269                                         logger.info("Skipping rundown {} {}", rundownID, rundownName);\r
270                                         continue;\r
271                                 }\r
272 \r
273                                 logger.info("Saving rundown {} {}", rundownID, rundownName);\r
274                                 copy(rundownArchive);\r
275                                 logger.info(systemMarker, "A '{}' tükör {}db bejátszójának archiválása sikeres volt", rundownArchive.getItemTitle(),\r
276                                                 rundownArchive.getStoryArchives().size());\r
277 \r
278                                 db.getCollection(ARCHIVEDRUNDOWNS).save(currentRundownID);\r
279                         } catch (Exception e) {\r
280                                 logger.catching(e);\r
281                                 logger.error(systemMarker,\r
282                                                 String.format("A %s %s tükör archiválása nem lehetséges, mert a annak ellenőrzése hibát jelzett. A rendszer üzenete: %s", rundownID,\r
283                                                                 rundownName, e.getMessage()));\r
284                         }\r
285                         index++;\r
286                 }\r
287 \r
288         }\r
289 \r
290         private StoryArchive processStory(BasicDBObject rundown, DBObject s) throws Exception {\r
291                 BasicDBObject story = (BasicDBObject) s;\r
292                 String parentStoryID = story.getString(IOctopusAPI.PARENT_STORY_ID);\r
293                 if (StringUtils.isBlank(parentStoryID)) {\r
294                         logger.warn("Story parentStoryID is null: {}", story.toPrettyString(null));\r
295                         return null;\r
296                 } else\r
297                         logger.debug("Processing story {}", parentStoryID);\r
298                 List<BasicDBObject> mosObjects = NoSQLUtils.asList(story, IOctopusAPI.MOS_OBJECTS);\r
299                 if (mosObjects == null)\r
300                         return null;\r
301                 StoryArchive storyArchive = null;\r
302                 for (BasicDBObject mosObject : mosObjects) {\r
303                         FileArchive fileArchive = processMosObject(rundown, story, mosObject);\r
304                         if (fileArchive == null)\r
305                                 continue;\r
306                         if (storyArchive == null) {\r
307                                 storyArchive = new StoryArchive();\r
308                                 storyArchive.setMediaHouseId(parentStoryID);\r
309                                 storyArchive.setMediaTitle(story.getString(IOctopusAPI.NAME));\r
310                                 storyArchive.setMediaDesc(story.getString(IOctopusAPI.SCRIPT_CONTENT));\r
311 \r
312                         }\r
313                         storyArchive.addFileArchive(fileArchive);\r
314                 }\r
315                 return storyArchive;\r
316 \r
317         }\r
318 \r
319         private List<BasicDBObject> queryArchivedRundowns() {\r
320                 List<BasicDBObject> result = null;\r
321                 DBCollection collection = db.getCollection(ARCHIVEDRUNDOWNS);\r
322                 DBCursor find = collection.find(new BasicDBObject(), new BasicDBObject(IOctopusAPI._ID, 0).append(IOctopusAPI.ID, 1));\r
323                 if (find.hasNext())\r
324                         result = ListUtils.cast(find.toArray());\r
325                 return result;\r
326         }\r
327 \r
328         private void setAndCheck(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,\r
329                         int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine) throws Exception {\r
330                 db = NoSQLUtils.getNoSQLDB();\r
331                 if (db == null) {\r
332                         logger.error(systemMarker, "Az NoSQL adatkezelő réteg nem elérhető.");\r
333                         throw new NullPointerException("Internal error, missing NoSQL DB reference.");\r
334                 }\r
335 \r
336                 if (jobEngine == null) {\r
337                         logger.error(systemMarker, "Az folyamatkezelő réteg nem elérhető.");\r
338                         throw new NullPointerException("Internal error, missing JobEngine reference.");\r
339                 }\r
340                 manager = jobEngine.getItemManager();\r
341                 if (manager == null) {\r
342                         logger.error(systemMarker, "Az adatbáziskezelő réteg nem elérhető.");\r
343                         throw new NullPointerException("Internal error, missing ItemManager reference.");\r
344                 }\r
345                 String nexioHost = System.getProperty("nexio.host");\r
346                 if (StringUtils.isBlank(nexioHost)) {\r
347                         logger.error(systemMarker, "A 'nexio.host' rendszer paraméter nem található.");\r
348                         throw new NullPointerException("System is not configured properly, 'jobengine.selenio.address' startup parameter missing.");\r
349                 }\r
350                 check(nexioPort, "nexioPort");\r
351                 check(nexioUserName, "nexioUserName");\r
352                 check(nexioPassword, "nexioPassword");\r
353 \r
354                 check(nexioKillDateDays, "nexioKillDateDays");\r
355                 this.nexioKillDateDays = nexioKillDateDays;\r
356                 check(nexioAgency, "nexioAgency");\r
357                 this.nexioAgency = nexioAgency;\r
358 \r
359                 sourceUri = manager.createStoreUri(RemoteStoreProtocol.FTP, nexioHost);\r
360                 sourceUri.setPortNumber(nexioPort);\r
361                 sourceUri.setUserName(nexioUserName);\r
362                 sourceUri.setPassword(nexioPassword);\r
363                 if (sourceUri == null) {\r
364                         logger.error(systemMarker, "A forrás nem elérhető.");\r
365                         throw new NullPointerException("Internal error, missing 'sourceUri'.");\r
366                 }\r
367 \r
368                 check(archiveFtp, "archiveFtp");\r
369                 check(archiveUserName, "archiveUserName");\r
370                 check(archivePassword, "archivePassword");\r
371 \r
372                 targetUri = manager.createStoreUri(new URI(archiveFtp));\r
373                 targetUri.setUserName(archiveUserName);\r
374                 targetUri.setPassword(archivePassword);\r
375                 if (targetUri == null) {\r
376                         logger.error(systemMarker, "A cél nem elérhető.");\r
377                         throw new NullPointerException("Internal error, missing 'targetUri'.");\r
378                 }\r
379 \r
380         }\r
381 \r
382         private void transferChunk(String fileName) throws Exception {\r
383                 logger.info("Transfer chunk {}", fileName);\r
384                 OutputStream outStream = null;\r
385                 try {\r
386                         targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
387                         outStream = targetFtp.storeFileStream(fileName);\r
388                         if (outStream == null) {\r
389                                 throw new NullPointerException("Can not open: " + fileName + MXFEXT + " Reply:" + targetFtp.getReplyString());\r
390                         }\r
391                 } catch (Exception e) {\r
392                         logger.catching(e);\r
393                         throw e;\r
394                 } finally {\r
395                         if (outStream != null)\r
396                                 outStream.close();\r
397                         targetUri.cleanUp();\r
398                 }\r
399         }\r
400 \r
401         private void transferFile(String sourceFileName, String targetFileName) throws Exception {\r
402                 int reply = 0;\r
403                 logger.info("Transfer clip {}", sourceFileName);\r
404                 try {\r
405                         sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();\r
406                         targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
407                         if (!targetFtp.enterRemotePassiveMode())\r
408                                 throw new Exception("!PASV");\r
409 \r
410                         reply = sourceFtp.port(InetAddress.getByName(targetFtp.getPassiveHost()), targetFtp.getPassivePort());\r
411                         if (!FTPReply.isPositiveCompletion(reply))\r
412                                 throw new Exception("PORT parancs válasza: " + sourceFtp.getReplyString());\r
413 \r
414                         if (!sourceFtp.setFileType(FTP.BINARY_FILE_TYPE))\r
415                                 throw new Exception("!SOURCE TYPE");\r
416 \r
417                         reply = sourceFtp.retr(sourceFileName);\r
418                         if (!FTPReply.isPositivePreliminary(reply))\r
419                                 throw new Exception("RETR parancs válasza: " + sourceFtp.getReplyString());\r
420 \r
421                         if (!targetFtp.setFileType(FTP.BINARY_FILE_TYPE))\r
422                                 throw new Exception("!TARGET TYPE");\r
423 \r
424                         reply = targetFtp.stor(targetFileName);\r
425                         if (!FTPReply.isPositivePreliminary(reply))\r
426                                 throw new Exception("STOR parancs válasza: " + sourceFtp.getReplyString());\r
427 \r
428                         while (true) {\r
429                                 reply = sourceFtp.stat();\r
430                                 if (!FTPReply.isPositiveCompletion(reply))\r
431                                         throw new Exception("STAT parancs válasza: " + sourceFtp.getReplyString());\r
432 \r
433                                 logger.info("Status: {}", sourceFtp.getReplyString());\r
434                                 if (reply == 226) {\r
435                                         break;\r
436                                 }\r
437                                 Thread.sleep(1000);\r
438                         }\r
439                 } catch (Exception e) {\r
440                         logger.catching(e);\r
441                         throw e;\r
442                 } finally {\r
443                         sourceUri.cleanUp();\r
444                         targetUri.cleanUp();\r
445                 }\r
446 \r
447         }\r
448 \r
449         private void transferMetadata(String fileName, BasicDBObject metadata) throws Exception {\r
450                 logger.info("Transfer metadata {}", fileName);\r
451                 OutputStream outStream = null;\r
452                 try {\r
453                         targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
454                         if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER)) {\r
455                                 targetFtp.makeDirectory(EscortFiles.STATUSFOLDER);\r
456                                 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER))\r
457                                         throw new Exception("!STATUSFOLDER");\r
458                         }\r
459 \r
460                         outStream = targetFtp.storeFileStream(fileName + JSON_EXT);\r
461                         if (outStream == null) {\r
462                                 throw new NullPointerException("Can not open: " + fileName + JSON_EXT + " Reply:" + targetFtp.getReplyString());\r
463                         }\r
464                         outStream.write(metadata.toString().getBytes(UTF_8));\r
465                         outStream.flush();\r
466                         //targetFtp.changeToParentDirectory();\r
467                 } catch (Exception e) {\r
468                         logger.catching(e);\r
469                         throw e;\r
470                 } finally {\r
471                         if (outStream != null)\r
472                                 outStream.close();\r
473                         targetUri.cleanUp();\r
474                 }\r
475         }\r
476 \r
477 }\r